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, 38 insertions, 12 deletions

diff --git a/ext/cache/lib.rs b/ext/cache/lib.rs
index b5a29e905..1459af8cb 100644
--- a/ext/cache/lib.rs
+++ b/ext/cache/lib.rs
@@ -14,6 +14,7 @@ use deno_core::ByteString;
 use deno_core::OpState;
 use deno_core::Resource;
 use deno_core::ResourceId;
+
 mod sqlite;
 pub use sqlite::SqliteBackedCache;
 
@@ -28,6 +29,7 @@ deno_core::extension!(deno_cache,
     op_cache_storage_has<CA>,
     op_cache_storage_delete<CA>,
     op_cache_put<CA>,
+    op_cache_put_finish<CA>,
     op_cache_match<CA>,
     op_cache_delete<CA>,
   ],
@@ -86,16 +88,24 @@ pub struct CacheDeleteRequest {
   pub request_url: String,
 }
 
-#[async_trait]
-pub trait Cache: Clone {
+#[async_trait(?Send)]
+pub trait Cache: Clone + 'static {
+  type CachePutResourceType: Resource;
+
   async fn storage_open(&self, cache_name: String) -> Result<i64, AnyError>;
   async fn storage_has(&self, cache_name: String) -> Result<bool, AnyError>;
   async fn storage_delete(&self, cache_name: String) -> Result<bool, AnyError>;
 
-  async fn put(
+  /// Create a put request.
+  async fn put_create(
     &self,
     request_response: CachePutRequest,
-  ) -> Result<Option<Rc<dyn Resource>>, AnyError>;
+  ) -> Result<Option<Rc<Self::CachePutResourceType>>, AnyError>;
+  /// Complete a put request.
+  async fn put_finish(
+    &self,
+    resource: Rc<Self::CachePutResourceType>,
+  ) -> Result<(), AnyError>;
   async fn r#match(
     &self,
     request: CacheMatchRequest,
@@ -113,7 +123,7 @@ pub async fn op_cache_storage_open<CA>(
   cache_name: String,
 ) -> Result<i64, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
   cache.storage_open(cache_name).await
@@ -125,7 +135,7 @@ pub async fn op_cache_storage_has<CA>(
   cache_name: String,
 ) -> Result<bool, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
   cache.storage_has(cache_name).await
@@ -137,7 +147,7 @@ pub async fn op_cache_storage_delete<CA>(
   cache_name: String,
 ) -> Result<bool, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
   cache.storage_delete(cache_name).await
@@ -149,10 +159,10 @@ pub async fn op_cache_put<CA>(
   request_response: CachePutRequest,
 ) -> Result<Option<ResourceId>, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
-  match cache.put(request_response).await? {
+  match cache.put_create(request_response).await? {
     Some(resource) => {
       let rid = state.borrow_mut().resource_table.add_rc_dyn(resource);
       Ok(Some(rid))
@@ -162,12 +172,28 @@ where
 }
 
 #[op]
+pub async fn op_cache_put_finish<CA>(
+  state: Rc<RefCell<OpState>>,
+  rid: ResourceId,
+) -> Result<(), AnyError>
+where
+  CA: Cache,
+{
+  let cache = get_cache::<CA>(&state)?;
+  let resource = state
+    .borrow_mut()
+    .resource_table
+    .get::<CA::CachePutResourceType>(rid)?;
+  cache.put_finish(resource).await
+}
+
+#[op]
 pub async fn op_cache_match<CA>(
   state: Rc<RefCell<OpState>>,
   request: CacheMatchRequest,
 ) -> Result<Option<CacheMatchResponse>, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
   match cache.r#match(request).await? {
@@ -186,7 +212,7 @@ pub async fn op_cache_delete<CA>(
   request: CacheDeleteRequest,
 ) -> Result<bool, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let cache = get_cache::<CA>(&state)?;
   cache.delete(request).await
@@ -194,7 +220,7 @@ where
 
 pub fn get_cache<CA>(state: &Rc<RefCell<OpState>>) -> Result<CA, AnyError>
 where
-  CA: Cache + 'static,
+  CA: Cache,
 {
   let mut state = state.borrow_mut();
   if let Some(cache) = state.try_borrow::<CA>() {