refactor(tvix/castore): drop is_closed() from impl DirectoryPutter

This is only used in the gRPC version (GRPCPutter), during the test
automation.

So define it as a method there, behind #[cfg(test)], and remove from
the trait.

Change-Id: Idf170884e3a10be0e96c75d946d9c431171e5e88
Reviewed-on: https://cl.tvl.fyi/c/depot/+/10340
Tested-by: BuildkiteCI
Autosubmit: flokli <flokli@flokli.de>
Reviewed-by: raitobezarius <tvl@lahfa.xyz>
This commit is contained in:
Florian Klink 2023-12-12 21:25:50 +02:00 committed by clbot
parent 36cc7b7088
commit 923a5737e6
3 changed files with 33 additions and 20 deletions

View file

@ -228,6 +228,16 @@ impl GRPCPutter {
rq: Some((task, directory_sender)), rq: Some((task, directory_sender)),
} }
} }
// allows checking if the tx part of the channel is closed.
// only used in the test case.
#[cfg(test)]
fn is_closed(&self) -> bool {
match self.rq {
None => true,
Some((_, ref directory_sender)) => directory_sender.is_closed(),
}
}
} }
#[async_trait] #[async_trait]
@ -272,28 +282,23 @@ impl DirectoryPutter for GRPCPutter {
} }
} }
} }
// allows checking if the tx part of the channel is closed.
fn is_closed(&self) -> bool {
match self.rq {
None => true,
Some((_, ref directory_sender)) => directory_sender.is_closed(),
}
}
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use core::time; use core::time;
use futures::StreamExt; use futures::StreamExt;
use std::{sync::Arc, time::Duration}; use std::{any::Any, sync::Arc, time::Duration};
use tempfile::TempDir; use tempfile::TempDir;
use tokio::net::UnixListener; use tokio::net::UnixListener;
use tokio_retry::{strategy::ExponentialBackoff, Retry}; use tokio_retry::{strategy::ExponentialBackoff, Retry};
use tokio_stream::wrappers::UnixListenerStream; use tokio_stream::wrappers::UnixListenerStream;
use crate::{ use crate::{
directoryservice::{DirectoryService, GRPCDirectoryService, MemoryDirectoryService}, directoryservice::{
grpc::GRPCPutter, DirectoryPutter, DirectoryService, GRPCDirectoryService,
MemoryDirectoryService,
},
fixtures::{self, DIRECTORY_A, DIRECTORY_B}, fixtures::{self, DIRECTORY_A, DIRECTORY_B},
proto::{directory_service_client::DirectoryServiceClient, GRPCDirectoryServiceWrapper}, proto::{directory_service_client::DirectoryServiceClient, GRPCDirectoryServiceWrapper},
utils::gen_directorysvc_grpc_client, utils::gen_directorysvc_grpc_client,
@ -400,6 +405,23 @@ mod tests {
let mut handle = directory_service.put_multiple_start(); let mut handle = directory_service.put_multiple_start();
handle.put(DIRECTORY_B.clone()).await.expect("must succeed"); handle.put(DIRECTORY_B.clone()).await.expect("must succeed");
// get a GRPCPutter, so we can peek at [is_closed].
let handle_any = &mut handle as &mut dyn Any;
// `unchecked_downcast_mut` is unstable for now,
// https://github.com/rust-lang/rust/issues/90850
// We do the same thing here.
// The reason for why we cannot use the checked downcast lies
// in the fact that:
// - GRPCPutter has type ID A
// - Box<GRPCPutter> has type ID B
// - "Box<dyn GRPCPutter>" (invalid type) has type ID C
// B seems different from C in this context.
// We cannot unpack and perform upcast coercion of the traits as it's an unstable
// feature.
// We cannot add `as_any` in `DirectoryPutter` as that would defeat the whole purpose
// of not making leak `is_closed` in the original trait.
let handle = unsafe { &mut *(handle_any as *mut dyn Any as *mut Box<GRPCPutter>) };
let mut is_closed = false; let mut is_closed = false;
for _try in 1..1000 { for _try in 1..1000 {
if handle.is_closed() { if handle.is_closed() {

View file

@ -73,8 +73,4 @@ pub trait DirectoryPutter: Send {
/// If there's been any invalid Directory message uploaded, and error *must* /// If there's been any invalid Directory message uploaded, and error *must*
/// be returned. /// be returned.
async fn close(&mut self) -> Result<B3Digest, Error>; async fn close(&mut self) -> Result<B3Digest, Error>;
/// Return whether the stream is closed or not.
/// Used from some [DirectoryService] implementations only.
fn is_closed(&self) -> bool;
} }

View file

@ -103,7 +103,7 @@ impl<DS: DirectoryService> SimplePutter<DS> {
} }
#[async_trait] #[async_trait]
impl<DS: DirectoryService> DirectoryPutter for SimplePutter<DS> { impl<DS: DirectoryService + 'static> DirectoryPutter for SimplePutter<DS> {
async fn put(&mut self, directory: proto::Directory) -> Result<(), Error> { async fn put(&mut self, directory: proto::Directory) -> Result<(), Error> {
if self.closed { if self.closed {
return Err(Error::StorageError("already closed".to_string())); return Err(Error::StorageError("already closed".to_string()));
@ -117,7 +117,6 @@ impl<DS: DirectoryService> DirectoryPutter for SimplePutter<DS> {
Ok(()) Ok(())
} }
/// We need to be mutable here, as that's the signature of the trait.
async fn close(&mut self) -> Result<B3Digest, Error> { async fn close(&mut self) -> Result<B3Digest, Error> {
if self.closed { if self.closed {
return Err(Error::StorageError("already closed".to_string())); return Err(Error::StorageError("already closed".to_string()));
@ -133,8 +132,4 @@ impl<DS: DirectoryService> DirectoryPutter for SimplePutter<DS> {
)), )),
} }
} }
fn is_closed(&self) -> bool {
self.closed
}
} }