test(service): Add tombstone inconsistency tests (#309)

Add tests for edge cases in the two-tier tombstone system that aren't
covered by the existing happy-path test:

- **Orphan cleanup on failed tombstone write**: If the long-term write
  succeeds but the subsequent tombstone write to high-volume fails, the
  long-term object must be cleaned up. Uses a `FailingPutBackend` to
  simulate the failure.
- **Orphan tombstone on get/get_metadata**: If a tombstone exists in
  high-volume but the corresponding long-term object is missing (e.g.
  partial cleanup or race), reads should return `None` gracefully rather
  than error or expose the tombstone.

Also extracts `from_backends` out of `#[cfg(test)]` so
`StorageService::new`
delegates to it, removing duplicated construction logic.

Documents the consistency invariant in the service architecture docs:
the tombstone is always the last thing written and the last thing
removed,
ensuring no orphans without distributed locks.

Author: jan-auer

objectstore-service/docs/architecture.md

@@ -46,6 +46,21 @@ A redirect tombstone is an empty object with
 in its metadata. It acts as a signpost: "the real data lives in the other
 backend."
 
+### Consistency Without Locks
+
+The tombstone system maintains consistency through operation ordering rather
+than distributed locks. The invariant is: a redirect tombstone is always the
+**last thing written** and the **last thing removed**.
+
+- On **write**, the real object is persisted before the tombstone. If the
+  tombstone write fails, the real object is rolled back.
+- On **delete**, the real object is removed before the tombstone. If the
+  long-term delete fails, the tombstone remains and the data stays reachable.
+
+This ensures that at every intermediate step, either the data is fully
+reachable (tombstone points to data) or fully absent — never an orphan in
+either direction.
+
 ### How Each Operation Handles Tombstones
 
 **Read** ([`StorageService::get_object`], [`StorageService::get_metadata`]):

objectstore-service/src/lib.rs

@@ -102,12 +102,14 @@ impl StorageService {
     ) -> anyhow::Result<Self> {
         let high_volume_backend = create_backend(high_volume_config).await?;
         let long_term_backend = create_backend(long_term_config).await?;
+        Ok(Self::from_backends(high_volume_backend, long_term_backend))
+    }
 
-        let inner = StorageServiceInner {
+    fn from_backends(high_volume_backend: BoxedBackend, long_term_backend: BoxedBackend) -> Self {
+        Self(Arc::new(StorageServiceInner {
             high_volume_backend,
             long_term_backend,
-        };
-        Ok(Self(Arc::new(inner)))
+        }))
     }
 
     /// Creates or overwrites an object.
@@ -430,6 +432,151 @@ mod tests {
         assert_eq!(file_contents.as_ref(), b"oh hai!");
     }
 
+    fn make_localfs_service() -> (StorageService, tempfile::TempDir, tempfile::TempDir) {
+        let hv_dir = tempfile::tempdir().unwrap();
+        let lt_dir = tempfile::tempdir().unwrap();
+        let hv = Box::new(backend::local_fs::LocalFsBackend::new(hv_dir.path()));
+        let lt = Box::new(backend::local_fs::LocalFsBackend::new(lt_dir.path()));
+        (StorageService::from_backends(hv, lt), hv_dir, lt_dir)
+    }
+
+    // --- Tombstone inconsistency tests ---
+
+    /// A backend where put_object always fails, but reads/deletes work normally.
+    #[derive(Debug)]
+    struct FailingPutBackend(backend::local_fs::LocalFsBackend);
+
+    #[async_trait::async_trait]
+    impl backend::common::Backend for FailingPutBackend {
+        fn name(&self) -> &'static str {
+            "failing-put"
+        }
+
+        async fn put_object(
+            &self,
+            _id: &ObjectId,
+            _metadata: &Metadata,
+            _stream: PayloadStream,
+        ) -> ServiceResult<()> {
+            Err(ServiceError::Io(std::io::Error::new(
+                std::io::ErrorKind::ConnectionRefused,
+                "simulated tombstone write failure",
+            )))
+        }
+
+        async fn get_object(
+            &self,
+            id: &ObjectId,
+        ) -> ServiceResult<Option<(Metadata, PayloadStream)>> {
+            self.0.get_object(id).await
+        }
+
+        async fn delete_object(&self, id: &ObjectId) -> ServiceResult<()> {
+            self.0.delete_object(id).await
+        }
+    }
+
+    /// If the tombstone write to the high-volume backend fails after the long-term
+    /// write succeeds, the long-term object must be cleaned up so we never leave
+    /// an unreachable orphan in long-term storage.
+    #[tokio::test]
+    async fn no_orphan_when_tombstone_write_fails() {
+        let lt_dir = tempfile::tempdir().unwrap();
+        let lt_backend_for_inspection = backend::local_fs::LocalFsBackend::new(lt_dir.path());
+
+        // High-volume backend always fails on put (simulating BigTable being down).
+        // This means the tombstone write will fail after the long-term write succeeds.
+        let hv: BoxedBackend = Box::new(FailingPutBackend(backend::local_fs::LocalFsBackend::new(
+            tempfile::tempdir().unwrap().path(),
+        )));
+        let lt: BoxedBackend = Box::new(backend::local_fs::LocalFsBackend::new(lt_dir.path()));
+        let service = StorageService::from_backends(hv, lt);
+
+        let payload = vec![0xABu8; 2 * 1024 * 1024]; // 2 MiB -> long-term path
+        let result = service
+            .insert_object(
+                make_context(),
+                Some("orphan-test".into()),
+                &Default::default(),
+                make_stream(&payload),
+            )
+            .await;
+
+        // The insert should fail (tombstone write failed)
+        assert!(result.is_err());
+
+        // The long-term object must have been cleaned up — no orphan
+        let id = ObjectId::from_parts(
+            "testing".into(),
+            Scopes::from_iter([Scope::create("testing", "value").unwrap()]),
+            "orphan-test".into(),
+        );
+        let orphan = lt_backend_for_inspection.get_object(&id).await.unwrap();
+        assert!(
+            orphan.is_none(),
+            "long-term object was not cleaned up after tombstone write failure"
+        );
+    }
+
+    /// If a tombstone exists in high-volume but the corresponding object is
+    /// missing from long-term storage (e.g. due to a race condition or partial
+    /// cleanup), reads should gracefully return None rather than error.
+    #[tokio::test]
+    async fn orphan_tombstone_returns_none_on_get() {
+        let (service, _hv_dir, lt_dir) = make_localfs_service();
+        let payload = vec![0xCDu8; 2 * 1024 * 1024]; // 2 MiB
+
+        let id = service
+            .insert_object(
+                make_context(),
+                Some("orphan-tombstone".into()),
+                &Default::default(),
+                make_stream(&payload),
+            )
+            .await
+            .unwrap();
+
+        // Manually delete the long-term object, leaving an orphan tombstone
+        let lt_backend = backend::local_fs::LocalFsBackend::new(lt_dir.path());
+        lt_backend.delete_object(&id).await.unwrap();
+
+        // get_object should gracefully return None, not error
+        let result = service.get_object(&id).await.unwrap();
+        assert!(
+            result.is_none(),
+            "orphan tombstone should resolve to None, not return the tombstone"
+        );
+    }
+
+    /// Same as above but for get_metadata — an orphan tombstone should return
+    /// None rather than exposing the tombstone metadata to callers.
+    #[tokio::test]
+    async fn orphan_tombstone_returns_none_on_get_metadata() {
+        let (service, _hv_dir, lt_dir) = make_localfs_service();
+        let payload = vec![0xEFu8; 2 * 1024 * 1024]; // 2 MiB
+
+        let id = service
+            .insert_object(
+                make_context(),
+                Some("orphan-tombstone-meta".into()),
+                &Default::default(),
+                make_stream(&payload),
+            )
+            .await
+            .unwrap();
+
+        // Manually delete the long-term object
+        let lt_backend = backend::local_fs::LocalFsBackend::new(lt_dir.path());
+        lt_backend.delete_object(&id).await.unwrap();
+
+        // get_metadata should gracefully return None
+        let result = service.get_metadata(&id).await.unwrap();
+        assert!(
+            result.is_none(),
+            "orphan tombstone metadata should resolve to None"
+        );
+    }
+
     #[tokio::test]
     async fn test_tombstone_redirect_and_delete() {
         let high_volume = StorageConfig::BigTable {