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.

3 files changed, 68 insertions, 31 deletions

diff --git a/ext/cache/01_cache.js b/ext/cache/01_cache.js
index 9476420ef..2ed8a50a2 100644
--- a/ext/cache/01_cache.js
+++ b/ext/cache/01_cache.js
@@ -1,5 +1,5 @@
 // Copyright 2018-2023 the Deno authors. All rights reserved. MIT license.
-
+// deno-lint-ignore-file camelcase
 const core = globalThis.Deno.core;
 import * as webidl from "ext:deno_webidl/00_webidl.js";
 const primordials = globalThis.__bootstrap.primordials;
@@ -20,7 +20,15 @@ import { toInnerResponse } from "ext:deno_fetch/23_response.js";
 import { URLPrototype } from "ext:deno_url/00_url.js";
 import { getHeader } from "ext:deno_fetch/20_headers.js";
 import { readableStreamForRid } from "ext:deno_web/06_streams.js";
-
+const {
+  op_cache_delete,
+  op_cache_match,
+  op_cache_put,
+  op_cache_put_finish,
+  op_cache_storage_delete,
+  op_cache_storage_has,
+  op_cache_storage_open,
+} = core.ensureFastOps();
 class CacheStorage {
   constructor() {
     webidl.illegalConstructor();
@@ -31,7 +39,7 @@ class CacheStorage {
     const prefix = "Failed to execute 'open' on 'CacheStorage'";
     webidl.requiredArguments(arguments.length, 1, prefix);
     cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
-    const cacheId = await core.opAsync("op_cache_storage_open", cacheName);
+    const cacheId = await op_cache_storage_open(cacheName);
     const cache = webidl.createBranded(Cache);
     cache[_id] = cacheId;
     return cache;
@@ -42,7 +50,7 @@ class CacheStorage {
     const prefix = "Failed to execute 'has' on 'CacheStorage'";
     webidl.requiredArguments(arguments.length, 1, prefix);
     cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
-    return await core.opAsync("op_cache_storage_has", cacheName);
+    return await op_cache_storage_has(cacheName);
   }
 
   async delete(cacheName) {
@@ -50,7 +58,7 @@ class CacheStorage {
     const prefix = "Failed to execute 'delete' on 'CacheStorage'";
     webidl.requiredArguments(arguments.length, 1, prefix);
     cacheName = webidl.converters["DOMString"](cacheName, prefix, "Argument 1");
-    return await core.opAsync("op_cache_storage_delete", cacheName);
+    return await op_cache_storage_delete(cacheName);
   }
 }
 
@@ -124,8 +132,7 @@ class Cache {
     reqUrl.hash = "";
 
     // Step 9-11.
-    const rid = await core.opAsync(
-      "op_cache_put",
+    const rid = await op_cache_put(
       {
         cacheId: this[_id],
         // deno-lint-ignore prefer-primordials
@@ -142,7 +149,7 @@ class Cache {
         while (true) {
           const { value, done } = await reader.read();
           if (done) {
-            await core.shutdown(rid);
+            await op_cache_put_finish(rid);
             break;
           }
           await core.writeAll(rid, value);
@@ -196,7 +203,7 @@ class Cache {
     ) {
       r = new Request(request);
     }
-    return await core.opAsync("op_cache_delete", {
+    return await op_cache_delete({
       cacheId: this[_id],
       requestUrl: r.url,
     });
@@ -240,8 +247,7 @@ class Cache {
       const url = new URL(r.url);
       url.hash = "";
       const innerRequest = toInnerRequest(r);
-      const matchResult = await core.opAsync(
-        "op_cache_match",
+      const matchResult = await op_cache_match(
         {
           cacheId: this[_id],
           // deno-lint-ignore prefer-primordials
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>() {
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 {