1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-11-28 16:20:57 -05:00

fix(unstable): kv watch should stop when db is closed (#21665)

Fixes #21634.
This commit is contained in:
Heyang Zhou 2023-12-22 05:04:17 +08:00 committed by Bartek Iwańczuk
parent 68254ddc20
commit b30dc11c4f
No known key found for this signature in database
GPG key ID: 0C6BCDDC3B3AD750
2 changed files with 30 additions and 5 deletions

View file

@ -2248,3 +2248,23 @@ dbTest("set with key versionstamp suffix", async (db) => {
"expected string, number, bigint, ArrayBufferView, boolean", "expected string, number, bigint, ArrayBufferView, boolean",
); );
}); });
Deno.test({
name: "watch should stop when db closed",
async fn() {
const db = await Deno.openKv(":memory:");
const watch = db.watch([["a"]]);
const completion = (async () => {
for await (const _item of watch) {
// pass
}
})();
setTimeout(() => {
db.close();
}, 100);
await completion;
},
});

View file

@ -444,16 +444,21 @@ async fn op_kv_watch_next(
let cancel_handle = resource.cancel_handle.clone(); let cancel_handle = resource.cancel_handle.clone();
let stream = RcRef::map(resource, |r| &r.stream) let stream = RcRef::map(resource, |r| &r.stream)
.borrow_mut() .borrow_mut()
.or_cancel(db_cancel_handle) .or_cancel(db_cancel_handle.clone())
.or_cancel(cancel_handle) .or_cancel(cancel_handle.clone())
.await; .await;
let Ok(Ok(mut stream)) = stream else { let Ok(Ok(mut stream)) = stream else {
return Ok(None); return Ok(None);
}; };
// doesn't need a cancel handle because the stream ends when the database // We hold a strong reference to `resource`, so we can't rely on the stream
// connection is closed // being dropped when the db connection is closed
let Some(res) = stream.next().await else { let Ok(Ok(Some(res))) = stream
.next()
.or_cancel(db_cancel_handle)
.or_cancel(cancel_handle)
.await
else {
return Ok(None); return Ok(None);
}; };