refactor(ext/cache): Remove custom shutdown and use fast async ops (#20107)

The original implementation of `Cache` used a custom `shutdown` method
on the resource, but to simplify fast streams work we're going to move
this to an op of its own.

While we're in here, we're going to replace `opAsync` with
`ensureFastOps`. `op2` work will have to wait because of some
limitations to our async support, however.

1 files changed, 13 insertions, 8 deletions

diff --git a/ext/cache/sqlite.rs b/ext/cache/sqlite.rs
index 4eb9924c7..8589d61fd 100644
--- a/ext/cache/sqlite.rs
+++ b/ext/cache/sqlite.rs
@@ -92,8 +92,10 @@ impl SqliteBackedCache {
   }
 }
 
-#[async_trait]
+#[async_trait(?Send)]
 impl Cache for SqliteBackedCache {
+  type CachePutResourceType = CachePutResource;
+
   /// Open a cache storage. Internally, this creates a row in the
   /// sqlite db if the cache doesn't exist and returns the internal id
   /// of the cache.
@@ -167,10 +169,10 @@ impl Cache for SqliteBackedCache {
     .await?
   }
 
-  async fn put(
+  async fn put_create(
     &self,
     request_response: CachePutRequest,
-  ) -> Result<Option<Rc<dyn Resource>>, AnyError> {
+  ) -> Result<Option<Rc<CachePutResource>>, AnyError> {
     let db = self.connection.clone();
     let cache_storage_dir = self.cache_storage_dir.clone();
     let now = SystemTime::now().duration_since(UNIX_EPOCH)?;
@@ -202,6 +204,13 @@ impl Cache for SqliteBackedCache {
     }
   }
 
+  async fn put_finish(
+    &self,
+    resource: Rc<CachePutResource>,
+  ) -> Result<(), AnyError> {
+    resource.write_to_cache().await
+  }
+
   async fn r#match(
     &self,
     request: CacheMatchRequest,
@@ -346,7 +355,7 @@ impl CachePutResource {
     Ok(data.len())
   }
 
-  async fn shutdown(self: Rc<Self>) -> Result<(), AnyError> {
+  async fn write_to_cache(self: Rc<Self>) -> Result<(), AnyError> {
     let resource = deno_core::RcRef::map(&self, |r| &r.file);
     let mut file = resource.borrow_mut().await;
     file.flush().await?;
@@ -377,10 +386,6 @@ impl Resource for CachePutResource {
   }
 
   deno_core::impl_writable!();
-
-  fn shutdown(self: Rc<Self>) -> AsyncResult<()> {
-    Box::pin(self.shutdown())
-  }
 }
 
 pub struct CacheResponseResource {