refactor(castore/directory): separate order logic from ClosureValidator

ClosureValidator was previously only suitable for a very narrow use case:
Validating incoming uploads, which are in leaves-to-root order.
This is because the ordering validation was hard-wired into the add()
function.
This
- Re-name ClosureValidator to DirectoryGraph, which is more suitable
  since it actually stores the Directory structs and is drained in the end.
- Move the ordering-related logic to a separate OrderValidator, which
  can be used independently.
- re-write DirectoryGraph to be a general purpose
  validator which can accept the input in both orders
  and can be drained in both orders as well.

This means the DirectoryGraph and OrderValidator can now serve
multiple new purposes:
- Validating the incoming closure on the client while downloading.
- Validating the incoming closure downloaded in a caching layer from the
  `far` cache, and re-order it for insertion into the `near` cache.

Change-Id: I2b4b226348416912d7a31935bec050e53d911b70
Reviewed-on: https://cl.tvl.fyi/c/depot/+/11708
Tested-by: BuildkiteCI
Reviewed-by: flokli <flokli@flokli.de>
Autosubmit: yuka <yuka@yuka.dev>

tvix/castore/src/proto/grpc_directoryservice_wrapper.rs

@@ -1,4 +1,5 @@
-use crate::directoryservice::ClosureValidator;
+use crate::directoryservice::DirectoryGraph;
+use crate::directoryservice::LeavesToRootValidator;
 use crate::proto;
 use crate::{directoryservice::DirectoryService, B3Digest};
 use futures::stream::BoxStream;
@@ -78,14 +79,20 @@ where
     ) -> Result<Response<proto::PutDirectoryResponse>, Status> {
         let mut req_inner = request.into_inner();
 
-        // We put all Directory messages we receive into ClosureValidator first.
-        let mut validator = ClosureValidator::default();
+        // We put all Directory messages we receive into DirectoryGraph.
+        let mut validator = DirectoryGraph::<LeavesToRootValidator>::default();
         while let Some(directory) = req_inner.message().await? {
-            validator.add(directory)?;
+            validator
+                .add(directory)
+                .map_err(|e| tonic::Status::new(tonic::Code::Internal, e.to_string()))?;
         }
 
         // drain, which validates connectivity too.
-        let directories = validator.finalize()?;
+        let directories = validator
+            .validate()
+            .map_err(|e| tonic::Status::new(tonic::Code::Internal, e.to_string()))?
+            .drain_leaves_to_root()
+            .collect::<Vec<_>>();
 
         let mut directory_putter = self.directory_service.put_multiple_start();
         for directory in directories {