majjo/Pumpkin
1
0
Fork 0
mirror of https://github.com/Pumpkin-MC/Pumpkin.git synced 2026-10-31 04:17:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(plugin): load v0.1 plugins on a sequential engine for upgrading to wasmtime 50.0.0 (#4098)

Browse Source 
* feat(plugin): load v0.1 plugins on a sequential engine and keep v0.2 concurrent

* chore(deps): move wasmtime off the git pin to 50.0.0

* fix(plugin): reject pump_blocking on a concurrent store before it runs

* refactor(plugin): clarify new error messages and comments
This commit is contained in:
Cayden Tisdale
2026-10-22 12:55:33 -04:00
committed by GitHub
parent 3cf12e890c
commit 9c04f714bb
13 changed files with 990 additions and 609 deletions
Download Patch File Download Diff File Expand all files Collapse all files

296
Cargo.lock generated
View File

@@ -73,12 +73,6 @@ version = "0.2.21"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
[[package]]
name = "ambient-authority"
version = "0.0.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e9d4ee0d472d1cd2e28c97dfa124b3d8d992e10eb0a035f33f5d12e3a177ba3b"
[[package]]
name = "anes"
version = "0.1.6"
@@ -462,24 +456,6 @@ dependencies = [
"serde",
]
[[package]]
name = "cap-primitives"
version = "4.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8b5f74729fd2f44701d1a8eb47e906cdb3ccd9ec0f02baad85a744b791940b18"
dependencies = [
"ambient-authority",
"fs-set-times",
"io-extras",
"io-lifetimes 3.0.1",
"ipnet",
"maybe-owned",
"rustix",
"rustix-linux-procfs",
"windows-sys 0.61.2",
"winx",
]
[[package]]
name = "cast"
version = "0.3.0"
@@ -812,24 +788,27 @@ dependencies = [
[[package]]
name = "cranelift-assembler-x64"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d3307701abcb90554ee72fdb0db36ee0334a4abe380c4d87a5b04a6942613c3a"
dependencies = [
"cranelift-assembler-x64-meta",
]
[[package]]
name = "cranelift-assembler-x64-meta"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e62d0e19f9f5fa0c4243236df1ffdf139d207600b4402ac41ceba8dbe7ca0f3d"
dependencies = [
"cranelift-srcgen",
]
[[package]]
name = "cranelift-bforest"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47784bc3c4b57c0182bb8ff04aae5d5f5fc6bae8ba26b1212c9a659b90dab84e"
dependencies = [
"cranelift-entity",
"wasmtime-internal-core",
@@ -837,8 +816,9 @@ dependencies = [
[[package]]
name = "cranelift-bitset"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e13d120406f713dbce1940751e77bbb14daad1261379b7200c19285953e79a1d"
dependencies = [
"serde",
"serde_derive",
@@ -847,8 +827,9 @@ dependencies = [
[[package]]
name = "cranelift-codegen"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a53dbec33ff86909e44be6f0ec2d17841aa38b931013707c9fc7279f1f0e5760"
dependencies = [
"bumpalo",
"cranelift-assembler-x64",
@@ -877,8 +858,9 @@ dependencies = [
[[package]]
name = "cranelift-codegen-meta"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "42295b38e9182954cb32a71812e3d64d66a90efc0a2ababb73e465ad4a1efccf"
dependencies = [
"cranelift-assembler-x64-meta",
"cranelift-codegen-shared",
@@ -889,21 +871,24 @@ dependencies = [
[[package]]
name = "cranelift-codegen-shared"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b5200fa877fb64cd30cabe0603f394333f88f69adf63e4db8d48a111d0c9d0dd"
[[package]]
name = "cranelift-control"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b86250e480795fbb4425ea703bca7b75282998458ce74f5360843a31949f2bbc"
dependencies = [
"arbitrary",
]
[[package]]
name = "cranelift-entity"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "903c6592c7d600c251d20929f143543b7303ac70b072ad3394d65b29caf56f69"
dependencies = [
"cranelift-bitset",
"serde",
@@ -913,8 +898,9 @@ dependencies = [
[[package]]
name = "cranelift-frontend"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b929f09b82a270e2aae062945cf86cd9db39c59ad9bead6986ea6b09fcf8a969"
dependencies = [
"cranelift-codegen",
"hashbrown 0.17.1",
@@ -925,23 +911,27 @@ dependencies = [
[[package]]
name = "cranelift-isle"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b1c5809e8636a37e5a25ecc35242d33d6e7391c49b67488b83446e2898af2a2f"
[[package]]
name = "cranelift-native"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "815c559daf605a5b4ea35530604643c1e87504353c0d1a14763f9c2b9d518493"
dependencies = [
"cranelift-codegen",
"libc",
"mach2",
"target-lexicon",
]
[[package]]
name = "cranelift-srcgen"
version = "0.136.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "0.137.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d633ff621e7f31a943431da0d7dd37c4376766c60bd7171216de1c7ccabfb370"
[[package]]
name = "crc"
@@ -1491,17 +1481,6 @@ dependencies = [
"percent-encoding",
]
[[package]]
name = "fs-set-times"
version = "0.20.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "94e7099f6313ecacbe1256e8ff9d617b75d1bcb16a6fddef94866d225a01a14a"
dependencies = [
"io-lifetimes 2.0.4",
"rustix",
"windows-sys 0.52.0",
]
[[package]]
name = "fsevent-sys"
version = "4.1.0"
@@ -2095,28 +2074,6 @@ dependencies = [
"hybrid-array",
]
[[package]]
name = "io-extras"
version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "20fd6de4ccfcc187e38bc21cfa543cb5a302cb86a8b114eb7f0bf0dc9f8ac00f"
dependencies = [
"io-lifetimes 3.0.1",
"windows-sys 0.60.2",
]
[[package]]
name = "io-lifetimes"
version = "2.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "06432fb54d3be7964ecd3649233cddf80db2832f47fec34c01f65b3d9d774983"
[[package]]
name = "io-lifetimes"
version = "3.0.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2f0fb0570afe1fed943c5c3d4102d5358592d8625fda6a0007fdbe65a92fba96"
[[package]]
name = "ipnet"
version = "2.12.2"
@@ -2359,12 +2316,6 @@ version = "0.8.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
[[package]]
name = "maybe-owned"
version = "0.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4facc753ae494aeb6e3c22f839b158aebd4f9270f55cd3c79906c45476c47ab4"
[[package]]
name = "md-5"
version = "0.10.6"
@@ -3061,8 +3012,9 @@ dependencies = [
[[package]]
name = "pulley-interpreter"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6af498fa3ca312fe5a07178757781059b75a331f24d53aeb3706e51ad73317bd"
dependencies = [
"cranelift-bitset",
"log",
@@ -3072,12 +3024,13 @@ dependencies = [
[[package]]
name = "pulley-macros"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0ba9959a3a286b7a2fb16466b7d1a804a117f81a98b7ac6d86cabd5ba037f3d"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
"syn 3.0.9",
]
[[package]]
@@ -3456,6 +3409,7 @@ dependencies = [
"tempfile",
"tokio",
"tracing",
"wasm-encoder 0.262.0",
"wasmtime",
"wasmtime-wasi",
"wasmtime-wasi-http",
@@ -5349,12 +5303,12 @@ dependencies = [
[[package]]
name = "wasm-encoder"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f64765e43c000fae008a48db8b288e175e237fc9ce20a26586836878c1bee89f"
checksum = "2608e8bb6d67fd68f5a8d0eb1363d6e7bcbc1f8ded5a0bd3a1e382462b876b22"
dependencies = [
"leb128fmt",
"wasmparser 0.258.3",
"wasmparser 0.261.0",
]
[[package]]
@@ -5369,14 +5323,14 @@ dependencies = [
[[package]]
name = "wasm-metadata"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "47e10808eb27efde692f1b1706089fe19296728bf872291f5b4e0d50fba88f2f"
checksum = "b88de21560ecd49deb8d8f1616b9b695f66fc312c5eca503f43accdc5795339e"
dependencies = [
"anyhow",
"indexmap",
"wasm-encoder 0.258.3",
"wasmparser 0.258.3",
"wasm-encoder 0.261.0",
"wasmparser 0.261.0",
]
[[package]]
@@ -5393,9 +5347,9 @@ dependencies = [
[[package]]
name = "wasmparser"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "fe9269dd817e0e7021bf534ccd9761bc2d64778b6708e41228b8439d75eaa0b1"
checksum = "4f20f20e44f7e8aeb6744823ea9d869ede51e51be4fdaedede2852282e54d2d8"
dependencies = [
"bitflags 2.13.2",
"hashbrown 0.17.1",
@@ -5418,19 +5372,20 @@ dependencies = [
[[package]]
name = "wasmprinter"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "14a457ba8670a3db3d357c1aeac3ed4fe378a76054a221789f0df0ea507e292b"
checksum = "86a21c9e3325655ba02de3e645bc8357a8b3651a2e3ada25b6dbc37560deb9b7"
dependencies = [
"anyhow",
"termcolor",
"wasmparser 0.258.3",
"wasmparser 0.261.0",
]
[[package]]
name = "wasmtime"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2d93539aae382faee7b19650fd966f1f8185599b9cc4adc0a5a8b8f768c11ba9"
dependencies = [
"addr2line",
"async-trait",
@@ -5454,7 +5409,7 @@ dependencies = [
"serde_derive",
"smallvec",
"target-lexicon",
"wasmparser 0.258.3",
"wasmparser 0.261.0",
"wasmtime-environ",
"wasmtime-internal-cache",
"wasmtime-internal-component-macro",
@@ -5471,8 +5426,9 @@ dependencies = [
[[package]]
name = "wasmtime-environ"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "34fd44c205ef29cc3c254253dceb916be586887602e0e380361ed022776cfc10"
dependencies = [
"anyhow",
"cpp_demangle",
@@ -5489,11 +5445,10 @@ dependencies = [
"semver",
"serde",
"serde_derive",
"sha2 0.10.9",
"smallvec",
"target-lexicon",
"wasm-encoder 0.258.3",
"wasmparser 0.258.3",
"wasm-encoder 0.261.0",
"wasmparser 0.261.0",
"wasmprinter",
"wasmtime-internal-component-util",
"wasmtime-internal-core",
@@ -5501,8 +5456,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-cache"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "067440fa60ddd5145840e0335ffe0ab4d908a021932b6844c3ad61570abddfef"
dependencies = [
"base64 0.22.1",
"directories-next",
@@ -5520,27 +5476,30 @@ dependencies = [
[[package]]
name = "wasmtime-internal-component-macro"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "adf0938f1a0f067a60aa07a7f4369fae05ef227314aca64c1abc7b2dc5e80743"
dependencies = [
"anyhow",
"proc-macro2",
"quote",
"syn 2.0.119",
"syn 3.0.9",
"wasmtime-internal-component-util",
"wasmtime-internal-wit-bindgen",
"wit-parser 0.258.3",
"wit-parser 0.261.0",
]
[[package]]
name = "wasmtime-internal-component-util"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2626398b21bf4b1547fdbadc68afd5cbe69c89ca2a1d6c45fd56fd3b28177b93"
[[package]]
name = "wasmtime-internal-core"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "83d4ac46034b149aed44335595402503094a04e3b70d0cff8c169d83437318b8"
dependencies = [
"hashbrown 0.17.1",
"libm",
@@ -5549,8 +5508,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-cranelift"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1f496bf29d3f2a1d060e0a497a20bfba2a36befe80bdf3e640ba8d3dcab7df35"
dependencies = [
"cranelift-codegen",
"cranelift-control",
@@ -5565,7 +5525,7 @@ dependencies = [
"smallvec",
"target-lexicon",
"thiserror 2.0.21",
"wasmparser 0.258.3",
"wasmparser 0.261.0",
"wasmtime-environ",
"wasmtime-internal-core",
"wasmtime-internal-unwinder",
@@ -5574,8 +5534,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-fiber"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2074dc297d5ec7f0cfff65c248798dfe4bbee4d760fc2956cc0eaf49d2eadef1"
dependencies = [
"cc",
"libc",
@@ -5587,8 +5548,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-jit-debug"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4e990e2fb7710dd73af071ecf60908072935be7a5c689b4a07e9a2adc7995151"
dependencies = [
"cc",
"wasmtime-internal-versioned-export-macros",
@@ -5596,8 +5558,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-jit-icache-coherence"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cf25a1b3b925915f845927967554400554db9cebd3a7038a6fca0cf6e9bbaf6a"
dependencies = [
"libc",
"wasmtime-internal-core",
@@ -5606,8 +5569,9 @@ dependencies = [
[[package]]
name = "wasmtime-internal-unwinder"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e69eb39fe5baf9e181269e6ced40f407d31afc8e514864e228ef0e087f4f5348"
dependencies = [
"cranelift-codegen",
"log",
@@ -5617,36 +5581,38 @@ dependencies = [
[[package]]
name = "wasmtime-internal-versioned-export-macros"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "a805a26effdf564820580b7c4ce1c00fdee3f4a9465a5aeef944f9e0c254ec2e"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
"syn 3.0.9",
]
[[package]]
name = "wasmtime-internal-wit-bindgen"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "8511353a609cd60bb91c85f96beef39a63e96ccdc20eb20af1da9b805c499ece"
dependencies = [
"anyhow",
"bitflags 2.13.2",
"heck",
"indexmap",
"wit-component 0.258.3",
"wit-parser 0.258.3",
"wit-component 0.261.0",
"wit-parser 0.261.0",
]
[[package]]
name = "wasmtime-wasi"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ed63d0061d45fdf32d8f47913845491395a6918e5b4bee51d1910c3777bcbaca"
dependencies = [
"async-trait",
"bitflags 2.13.2",
"bytes",
"cap-primitives",
"futures",
"rand 0.10.3",
"rustix",
@@ -5658,13 +5624,13 @@ dependencies = [
"wasmtime",
"wasmtime-wasi-io",
"windows-sys 0.61.2",
"winx",
]
[[package]]
name = "wasmtime-wasi-http"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cc344236952fbd212ce75fea430207783603669661f6634deff84dbc63ec13c0"
dependencies = [
"async-trait",
"bytes",
@@ -5674,6 +5640,7 @@ dependencies = [
"http-body-util",
"hyper",
"pin-project-lite",
"rustix",
"rustls",
"tokio",
"tokio-rustls",
@@ -5687,8 +5654,9 @@ dependencies = [
[[package]]
name = "wasmtime-wasi-io"
version = "49.0.0-dev"
source = "git+https://github.com/bytecodealliance/wasmtime?rev=f6b3140dbb1d9b436ef523d14740292c956b0364#f6b3140dbb1d9b436ef523d14740292c956b0364"
version = "50.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "273d9f9178b9415fba5a1b84d6634e82a1d95b5a0b657bd12d90a7d1073d27e0"
dependencies = [
"async-trait",
"bytes",
@@ -6046,16 +6014,6 @@ version = "1.0.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "23b97319f7b8343df12cc98938e5c3eb436064524c8d2b4e30a1d3a36eecdf81"
[[package]]
name = "winx"
version = "0.36.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3f3fd376f71958b862e7afb20cfe5a22830e1963462f3a17f49d82a6c1d1f42d"
dependencies = [
"bitflags 2.13.2",
"windows-sys 0.52.0",
]
[[package]]
name = "wit-bindgen"
version = "0.57.1"
@@ -6115,9 +6073,9 @@ dependencies = [
[[package]]
name = "wit-component"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "f0fca88950af0c1f42bc77e17315e48798b8d68340171b6d204f62bc19a878a6"
checksum = "2818766842bd2321074c348fb5c15d8fdb105a9eb78374e6d77c7d4127e1f2f9"
dependencies = [
"anyhow",
"bitflags 2.13.2",
@@ -6126,10 +6084,10 @@ dependencies = [
"serde",
"serde_derive",
"serde_json",
"wasm-encoder 0.258.3",
"wasm-metadata 0.258.3",
"wasmparser 0.258.3",
"wit-parser 0.258.3",
"wasm-encoder 0.261.0",
"wasm-metadata 0.261.0",
"wasmparser 0.261.0",
"wit-parser 0.261.0",
]
[[package]]
@@ -6165,9 +6123,9 @@ dependencies = [
[[package]]
name = "wit-parser"
version = "0.258.3"
version = "0.261.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "72a2127663dcf3ee16614fb0e3c33d5c6c66fec9dad317a8b0fdd3869487a533"
checksum = "962e43c4db2da607b50839a531e5b63473268b40efdf8028d5c54b572ca4f6f6"
dependencies = [
"anyhow",
"hashbrown 0.17.1",
@@ -6179,7 +6137,7 @@ dependencies = [
"serde_derive",
"serde_json",
"unicode-ident",
"wasmparser 0.258.3",
"wasmparser 0.261.0",
]
[[package]]

10
Cargo.toml
View File

@@ -263,11 +263,9 @@ rustc_version_runtime = { version = "0.3", default-features = false }
ordered-float = { version = "5.4", default-features = false, features = ["std"] }
xxhash-rust = { version = "0.8", default-features = false, features = ["xxh64"] }
# Temporary pin to the approved implementation of Component Model sync reentry:
# https://github.com/bytecodealliance/wasmtime/commit/f6b3140dbb1d9b436ef523d14740292c956b0364
wasmtime = { git = "https://github.com/bytecodealliance/wasmtime", rev = "f6b3140dbb1d9b436ef523d14740292c956b0364", default-features = false, features = ["runtime", "component-model", "component-model-async", "async", "cache", "cranelift", "gc", "gc-drc", "threads", "std"] }
wasmtime-wasi = { git = "https://github.com/bytecodealliance/wasmtime", rev = "f6b3140dbb1d9b436ef523d14740292c956b0364", default-features = false, features = ["p2", "p3"] }
wasmtime-wasi-http = { git = "https://github.com/bytecodealliance/wasmtime", rev = "f6b3140dbb1d9b436ef523d14740292c956b0364", default-features = false, features = ["p2", "p3", "default-send-request"] }
wasmtime = { version = "50.0.0", default-features = false, features = ["runtime", "component-model", "component-model-async", "async", "cache", "cranelift", "gc", "gc-drc", "threads", "std"] }
wasmtime-wasi = { version = "50.0.0", default-features = false, features = ["p2", "p3"] }
wasmtime-wasi-http = { version = "50.0.0", default-features = false, features = ["p2", "p3", "default-send-request"] }
# needed for interacting with wasmtime-wasi-http - keep in sync
hyper = { version = "^1.12", default-features = false }
@@ -279,7 +277,7 @@ wit-bindgen = { version = "0.63", default-features = false, features = ["macros"
postcard = { version = "1.1", default-features = false, features = ["alloc"] }
tracing-serde-structured = { version = "0.4", default-features = false }
wasmparser = { version = "0.262", default-features = false }
wasmparser = { version = "0.262", default-features = false, features = ["component-model", "std"] }
wasm-encoder = { version = "0.262", default-features = false, features = ["component-model", "std"] }
webpki-root-certs = { version = "1.0.10", default-features = false }
# ring-only: must match the provider the rest of the tree (wasmtime-wasi-http, rtc) uses

508
crates/pumpkin-plugin-runtime/src/executor.rs
View File

@@ -14,8 +14,8 @@ use std::{
use futures::{StreamExt, stream::FuturesUnordered};
use tokio::sync::{Mutex, mpsc, oneshot};
use wasmtime::{
AsContextMut, Store, StoreContextMut,
component::{Accessor, AccessorTask, ComponentNamedList, Lift, Lower, TypedFunc},
AsContext, AsContextMut, Store, StoreContextMut,
component::{Accessor, ComponentNamedList, Func, Lift, Lower, TypedFunc},
};
use crate::{
@@ -28,46 +28,61 @@ use crate::{
const STORE_QUEUE_CAPACITY: usize = 64;
const REENTRY_QUEUE_CAPACITY: usize = 64;
/// How a Store's guest calls are driven.
///
/// A Store takes its mode from the `Engine` it was created on, so both must agree.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub enum StoreMode {
/// One call at a time through `call_async`, on an engine without Wasmtime concurrency
/// support. A host import may re-enter the guest of the Store it is running on, which Wasmtime
/// rejects once concurrency support is on.
Sequential,
/// Calls run as tasks in Wasmtime's concurrent event loop, on an engine with concurrency
/// support. Host imports reach the guest through the event loop instead of nesting calls.
Concurrent,
}
pub type StoreFuture<'a, R> = Pin<Box<dyn Future<Output = wasmtime::Result<R>> + Send + 'a>>;
trait StoreJob<T>: Send {
fn run(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()>;
fn run_concurrent(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()>;
fn run_sequential(self: Box<Self>, store: StoreContextMut<'_, T>) -> StoreFuture<'_, ()>;
}
trait GuestStoreJob<T>: Send {
fn run_concurrent(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()>;
fn run_reentrant(self: Box<Self>, store: StoreContextMut<'_, T>) -> StoreFuture<'_, ()>;
fn run_sequential(
self: Box<Self>,
store: StoreContextMut<'_, T>,
reentrant: bool,
) -> StoreFuture<'_, ()>;
}
struct StoreCall<F, R> {
call: F,
result: oneshot::Sender<wasmtime::Result<R>>,
context: Option<ReentryContext>,
}
impl<T, F, R> StoreJob<T> for StoreCall<F, R>
where
T: Send + 'static,
F: for<'a> FnOnce(&'a Accessor<T>) -> StoreFuture<'a, R> + Send + 'static,
F: FnOnce(&mut T) -> R + Send + 'static,
R: Send + 'static,
{
fn run(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()> {
let Self {
call,
result,
context,
} = *self;
let future = call(accessor);
Box::pin(async move {
let output = if let Some(context) = context {
scope(context, future).await
} else {
future.await
};
let _ = result.send(output);
Ok(())
})
fn run_concurrent(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()> {
let Self { call, result } = *self;
let output = accessor.with(|mut access| call(access.data_mut()));
let _ = result.send(Ok(output));
Box::pin(std::future::ready(Ok(())))
}
fn run_sequential(self: Box<Self>, mut store: StoreContextMut<'_, T>) -> StoreFuture<'_, ()> {
let Self { call, result } = *self;
let output = call(store.data_mut());
let _ = result.send(Ok(output));
Box::pin(std::future::ready(Ok(())))
}
}
@@ -77,32 +92,60 @@ struct ShutdownStoreCall<F, R> {
context: ReentryContext,
}
impl<F, R> ShutdownStoreCall<F, R>
where
R: Send + 'static,
{
async fn finish(
context: ReentryContext,
result: oneshot::Sender<wasmtime::Result<R>>,
future: impl Future<Output = wasmtime::Result<R>>,
) -> wasmtime::Result<()> {
match scope(context, future).await {
Ok(value) => {
let _ = result.send(Ok(value));
Ok(())
}
Err(error) => {
let message = error.to_string();
let _ = result.send(Err(error));
Err(wasmtime::Error::msg(message))
}
}
}
}
impl<T, F, R> StoreJob<T> for ShutdownStoreCall<F, R>
where
T: Send + 'static,
F: for<'a> FnOnce(&'a Accessor<T>) -> StoreFuture<'a, R> + Send + 'static,
F: for<'a> FnOnce(LegacyGuestScope<'a, T>) -> StoreFuture<'a, R> + Send + 'static,
R: Send + 'static,
{
fn run(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()> {
fn run_concurrent(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()> {
let Self {
call,
result,
context,
} = *self;
let future = call(accessor);
Box::pin(async move {
match scope(context, future).await {
Ok(value) => {
let _ = result.send(Ok(value));
Ok(())
}
Err(error) => {
let message = error.to_string();
let _ = result.send(Err(error));
Err(wasmtime::Error::msg(message))
}
}
})
let future = call(LegacyGuestScope::concurrent(
accessor,
Arc::new(GuestCallFailure::default()),
));
Box::pin(Self::finish(context, result, future))
}
fn run_sequential(self: Box<Self>, store: StoreContextMut<'_, T>) -> StoreFuture<'_, ()> {
let Self {
call,
result,
context,
} = *self;
let future = call(LegacyGuestScope::sequential(
store,
Arc::new(GuestCallFailure::default()),
false,
));
Box::pin(Self::finish(context, result, future))
}
}
@@ -130,6 +173,46 @@ impl GuestCallFailure {
}
}
impl<T, F, R> GuestStoreCall<T, F, R>
where
T: Send + 'static,
F: for<'a> FnOnce(LegacyGuestScope<'a, T>) -> StoreFuture<'a, R> + Send + 'static,
R: Send + 'static,
{
async fn execute<'a>(
self,
make_scope: impl FnOnce(Arc<GuestCallFailure>) -> LegacyGuestScope<'a, T>,
) -> wasmtime::Result<()> {
let Self {
call,
result,
context,
reentry,
guest_call_failure,
root_admission,
} = self;
let failure_for_scope = Arc::clone(&guest_call_failure);
let output = scope(context, async move {
let _active_context = reentry.enter_active_context(context);
call(make_scope(failure_for_scope)).await
})
.await;
let guest_call_failure = guest_call_failure.message();
drop(root_admission);
if let Some(message) = guest_call_failure {
let output = match output {
Ok(_) => Err(wasmtime::Error::msg(message.clone())),
Err(error) => Err(error),
};
let _ = result.send(output);
Err(wasmtime::Error::msg(message))
} else {
let _ = result.send(output);
Ok(())
}
}
}
impl<T, F, R> GuestStoreJob<T> for GuestStoreCall<T, F, R>
where
T: Send + 'static,
@@ -137,73 +220,32 @@ where
R: Send + 'static,
{
fn run_concurrent(self: Box<Self>, accessor: &Accessor<T>) -> StoreFuture<'_, ()> {
let Self {
call,
result,
context,
reentry,
guest_call_failure,
root_admission,
} = *self;
Box::pin(async move {
let failure_for_scope = Arc::clone(&guest_call_failure);
let output = scope(context, async move {
let _active_context = reentry.enter_active_context(context);
call(LegacyGuestScope::concurrent(accessor, failure_for_scope)).await
})
.await;
let guest_call_failure = guest_call_failure.message();
drop(root_admission);
if let Some(message) = guest_call_failure {
let output = match output {
Ok(_) => Err(wasmtime::Error::msg(message.clone())),
Err(error) => Err(error),
};
let _ = result.send(output);
Err(wasmtime::Error::msg(message))
} else {
let _ = result.send(output);
Ok(())
}
})
Box::pin((*self).execute(move |failure| LegacyGuestScope::concurrent(accessor, failure)))
}
fn run_reentrant(self: Box<Self>, store: StoreContextMut<'_, T>) -> StoreFuture<'_, ()> {
let Self {
call,
result,
context,
reentry,
guest_call_failure,
root_admission,
} = *self;
Box::pin(async move {
let failure_for_scope = Arc::clone(&guest_call_failure);
let output = scope(context, async move {
let _active_context = reentry.enter_active_context(context);
call(LegacyGuestScope::reentrant(store, failure_for_scope)).await
})
.await;
let guest_call_failure = guest_call_failure.message();
drop(root_admission);
if let Some(message) = guest_call_failure {
let output = match output {
Ok(_) => Err(wasmtime::Error::msg(message.clone())),
Err(error) => Err(error),
};
let _ = result.send(output);
Err(wasmtime::Error::msg(message))
} else {
let _ = result.send(output);
Ok(())
}
})
fn run_sequential(
self: Box<Self>,
store: StoreContextMut<'_, T>,
reentrant: bool,
) -> StoreFuture<'_, ()> {
Box::pin(
(*self).execute(move |failure| LegacyGuestScope::sequential(store, failure, reentrant)),
)
}
}
enum LegacyGuestAccess<'a, T: 'static> {
Concurrent(&'a Accessor<T>),
Reentrant(StoreContextMut<'a, T>),
/// The Store is driven one call at a time. `reentrant` is set when the call was routed into a
/// host import that is already running on the Store.
Sequential {
store: StoreContextMut<'a, T>,
#[cfg_attr(
not(test),
expect(dead_code, reason = "tests assert which path routed the call")
)]
reentrant: bool,
},
}
/// Capability available while one synchronous legacy guest call is active.
@@ -229,19 +271,34 @@ where
}
}
const fn reentrant(
const fn sequential(
store: StoreContextMut<'a, T>,
guest_call_failure: Arc<GuestCallFailure>,
reentrant: bool,
) -> Self {
Self {
access: LegacyGuestAccess::Reentrant(store),
access: LegacyGuestAccess::Sequential { store, reentrant },
guest_call_failure,
}
}
#[cfg(test)]
pub(crate) const fn is_reentrant(&self) -> bool {
matches!(self.access, LegacyGuestAccess::Reentrant(_))
matches!(
self.access,
LegacyGuestAccess::Sequential {
reentrant: true,
..
}
)
}
#[cfg(test)]
pub(crate) const fn accessor(&self) -> Option<&'a Accessor<T>> {
match &self.access {
LegacyGuestAccess::Concurrent(accessor) => Some(*accessor),
LegacyGuestAccess::Sequential { .. } => None,
}
}
pub fn with<R>(&mut self, call: impl for<'store> FnOnce(StoreDataMut<'store, T>) -> R) -> R {
@@ -251,7 +308,7 @@ where
data: access.data_mut(),
})
}),
LegacyGuestAccess::Reentrant(store) => call(StoreDataMut {
LegacyGuestAccess::Sequential { store, .. } => call(StoreDataMut {
data: store.data_mut(),
}),
}
@@ -273,7 +330,7 @@ where
LegacyGuestAccess::Concurrent(accessor) => {
function.call_concurrent(&**accessor, params).await
}
LegacyGuestAccess::Reentrant(store) => {
LegacyGuestAccess::Sequential { store, .. } => {
function.call_async(store.as_context_mut(), params).await
}
};
@@ -282,6 +339,29 @@ where
}
result
}
/// Like [`Self::call`], for exports whose generated `TypedFunc` borrows its arguments and so
/// cannot be moved into a `'static` job. Takes the untyped `Func` and re-types it here with
/// owned arguments.
pub async fn call_func<Params, Return>(
&mut self,
function: Func,
params: Params,
) -> wasmtime::Result<Return>
where
Params: ComponentNamedList + Lower + 'static,
Return: ComponentNamedList + Lift + 'static,
{
let function = match &self.access {
LegacyGuestAccess::Concurrent(accessor) => {
accessor.with(|access| function.typed::<Params, Return>(access))?
}
LegacyGuestAccess::Sequential { store, .. } => {
function.typed::<Params, Return>(store.as_context())?
}
};
self.call(function, params).await
}
}
pub struct StoreDataMut<'a, T> {
@@ -294,28 +374,6 @@ impl<T> StoreDataMut<'_, T> {
}
}
struct ConcurrentTask<T>(Box<dyn StoreJob<T>>);
impl<T> AccessorTask<T> for ConcurrentTask<T>
where
T: Send + 'static,
{
fn run(self, accessor: &Accessor<T>) -> impl Future<Output = wasmtime::Result<()>> + Send {
self.0.run(accessor)
}
}
struct ConcurrentGuestTask<T>(Box<dyn GuestStoreJob<T>>);
impl<T> AccessorTask<T> for ConcurrentGuestTask<T>
where
T: Send + 'static,
{
fn run(self, accessor: &Accessor<T>) -> impl Future<Output = wasmtime::Result<()>> + Send {
self.0.run_concurrent(accessor)
}
}
enum StoreMessage<T> {
Call(Box<dyn StoreJob<T>>),
GuestCall(Box<dyn GuestStoreJob<T>>),
@@ -505,6 +563,7 @@ struct StoreShared<T, P> {
reentry: Arc<ReentryState<T>>,
policy: P,
spawner: Arc<dyn RuntimeSpawner>,
mode: StoreMode,
}
/// Cloneable submission and observation capability for one Wasmtime Store.
@@ -535,11 +594,15 @@ where
self.shared.lifecycle.subscribe()
}
/// Queues an owned Store operation. Once accepted, the operation runs even
/// if the caller stops waiting for its result.
/// Queues a short synchronous operation on the Store data. Once accepted,
/// the operation runs even if the caller stops waiting for its result.
///
/// The closure cannot await, so it cannot delay other plugin work beyond its own runtime. Keep
/// it brief; anything that needs the guest or an async host operation belongs in
/// [`StoreHandle::call_guest`].
pub async fn call<R, F>(&self, call: F) -> wasmtime::Result<R>
where
F: for<'a> FnOnce(&'a Accessor<T>) -> StoreFuture<'a, R> + Send + 'static,
F: FnOnce(&mut T) -> R + Send + 'static,
R: Send + 'static,
{
let (result, receiver) = oneshot::channel();
@@ -551,11 +614,7 @@ where
.reserve()
.await
.map_err(|_| self.terminal_error("Wasm plugin store driver is not running"))?;
permit.send(StoreMessage::Call(Box::new(StoreCall {
call,
result,
context: None,
})));
permit.send(StoreMessage::Call(Box::new(StoreCall { call, result })));
drop(send_guard);
self.receive_result(receiver, "Wasm plugin store call did not complete")
@@ -675,6 +734,10 @@ where
/// Awaits host work in the active guest call's chain with the reentry depth incremented
///
/// For [`StoreMode::Concurrent`] stores. A sequential store runs one guest call at a time, so
/// guest calls made from the awaited future would queue behind the active one; use
/// [`StoreHandle::pump_reentry`] there instead
///
/// Host imports are polled by Wasmtime's own event loop, outside the guest call's task-local
/// scope, so the context has to come from the Store's active-context registry
/// Without an active guest call, the future runs without adding a reentry context
@@ -738,10 +801,22 @@ where
S: AsContextMut<Data = T> + Send,
F: Future<Output = R> + Send,
{
self.ensure_sequential()?;
let context = self.next_reentry_context()?;
self.pump_reentry_with_context(store, context, future).await
}
/// Pumping only works when guest calls run one at a time. Checked before any work starts, so a
/// rejected call never runs its operation.
fn ensure_sequential(&self) -> wasmtime::Result<()> {
if self.shared.mode == StoreMode::Concurrent {
return Err(wasmtime::Error::msg(
"Reentrant callbacks can only be pumped on a sequential Wasm plugin store",
));
}
Ok(())
}
fn next_reentry_context(&self) -> wasmtime::Result<ReentryContext> {
self.shared
.reentry
@@ -770,7 +845,7 @@ where
let output = loop {
tokio::select! {
Some(job) = receiver.recv() => {
job.run_reentrant(store.as_context_mut()).await?;
job.run_sequential(store.as_context_mut(), true).await?;
}
output = &mut scoped_future => break output,
}
@@ -779,7 +854,7 @@ where
registration.close();
receiver.close();
while let Ok(job) = receiver.try_recv() {
job.run_reentrant(store.as_context_mut()).await?;
job.run_sequential(store.as_context_mut(), true).await?;
}
Ok(output)
}
@@ -792,6 +867,7 @@ where
S: AsContextMut<Data = T> + Send,
F: FnOnce() -> R + Send + 'static,
{
self.ensure_sequential()?;
let context = self.next_reentry_context()?;
let (result, receiver) = oneshot::channel();
self.shared
@@ -839,6 +915,7 @@ where
store: Store<T>,
policy: P,
spawner: Arc<dyn RuntimeSpawner>,
mode: StoreMode,
) -> wasmtime::Result<Self> {
let (sender, receiver) = mpsc::channel(STORE_QUEUE_CAPACITY);
let (control, control_receiver) = mpsc::unbounded_channel();
@@ -854,6 +931,7 @@ where
reentry: Arc::new(ReentryState::new()),
policy,
spawner: Arc::clone(&spawner),
mode,
});
let (ready_sender, ready) = oneshot::channel();
let policy_name = P::NAME;
@@ -874,6 +952,7 @@ where
Arc::clone(&task_guard.lifecycle),
ready_sender,
policy_name,
mode,
)
.await;
task_guard.disarm();
@@ -946,7 +1025,7 @@ where
/// drains Wasmtime tasks, and drops the Store.
pub async fn shutdown<R, F>(&self, call: F) -> wasmtime::Result<R>
where
F: for<'a> FnOnce(&'a Accessor<T>) -> StoreFuture<'a, R> + Send + 'static,
F: for<'a> FnOnce(LegacyGuestScope<'a, T>) -> StoreFuture<'a, R> + Send + 'static,
R: Send + 'static,
{
let (result, receiver) = oneshot::channel();
@@ -1060,8 +1139,135 @@ fn control_result(message: Option<&ControlMessage>) -> wasmtime::Result<()> {
}
}
#[allow(clippy::too_many_lines)]
async fn run_driver<T>(
store: Store<T>,
receiver: mpsc::Receiver<StoreMessage<T>>,
control: mpsc::UnboundedReceiver<ControlMessage>,
lifecycle: Arc<Lifecycle>,
ready: oneshot::Sender<()>,
policy_name: &'static str,
mode: StoreMode,
) where
T: Send + 'static,
{
match mode {
StoreMode::Concurrent => {
run_concurrent_driver(store, receiver, control, lifecycle, ready, policy_name).await;
}
StoreMode::Sequential => {
run_sequential_driver(store, receiver, control, lifecycle, ready, policy_name).await;
}
}
}
async fn run_sequential_driver<T>(
mut store: Store<T>,
mut receiver: mpsc::Receiver<StoreMessage<T>>,
mut control: mpsc::UnboundedReceiver<ControlMessage>,
lifecycle: Arc<Lifecycle>,
ready: oneshot::Sender<()>,
policy_name: &'static str,
) where
T: Send + 'static,
{
lifecycle.transition(DriverState::Accepting);
let _ = ready.send(());
let result = drive_store(&mut store, &mut receiver, &mut control, &lifecycle).await;
drop(store);
match result {
Ok(()) => lifecycle.transition(DriverState::Stopped),
Err(error) => {
let error = Arc::new(DriverError::new(format!("{error:#}")));
tracing::error!(
%error,
wasm_plugin_policy = policy_name,
"Wasm plugin store driver stopped"
);
lifecycle.transition(DriverState::Failed(error));
}
}
}
/// Outcome of handling one queued message.
enum Step {
Continue,
/// A shutdown job ran; the driver stops with its result.
Finished(wasmtime::Result<()>),
}
async fn run_sequential_message<T>(
store: &mut Store<T>,
message: StoreMessage<T>,
lifecycle: &Lifecycle,
) -> wasmtime::Result<Step>
where
T: Send + 'static,
{
match message {
StoreMessage::Call(job) => job.run_sequential(store.as_context_mut()).await?,
StoreMessage::GuestCall(job) => job.run_sequential(store.as_context_mut(), false).await?,
StoreMessage::Shutdown {
job,
root_admission,
} => {
lifecycle.transition(DriverState::Stopping);
let result = job.run_sequential(store.as_context_mut()).await;
drop(root_admission);
return Ok(Step::Finished(result));
}
}
Ok(Step::Continue)
}
async fn drive_store<T>(
store: &mut Store<T>,
receiver: &mut mpsc::Receiver<StoreMessage<T>>,
control: &mut mpsc::UnboundedReceiver<ControlMessage>,
lifecycle: &Lifecycle,
) -> wasmtime::Result<()>
where
T: Send + 'static,
{
loop {
let message = tokio::select! {
biased;
Some(control_message) = control.recv() => {
lifecycle.transition(DriverState::Draining);
receiver.close();
while let Some(message) = receiver.recv().await {
if let Step::Finished(result) =
run_sequential_message(store, message, lifecycle).await?
{
return result;
}
}
lifecycle.transition(DriverState::Stopping);
return control_result(Some(&control_message));
}
message = receiver.recv() => message,
};
let Some(message) = message else {
lifecycle.transition(DriverState::Draining);
// The owner may have closed the submission channel while its control signal is still pending
let control_message = control.recv().await;
lifecycle.transition(DriverState::Stopping);
return control_result(control_message.as_ref());
};
if matches!(message, StoreMessage::Shutdown { .. }) {
lifecycle.transition(DriverState::Draining);
receiver.close();
}
if let Step::Finished(result) = run_sequential_message(store, message, lifecycle).await? {
return result;
}
}
}
#[allow(clippy::too_many_lines)]
async fn run_concurrent_driver<T>(
mut store: Store<T>,
mut receiver: mpsc::Receiver<StoreMessage<T>>,
mut control: mpsc::UnboundedReceiver<ControlMessage>,
@@ -1099,10 +1305,10 @@ async fn run_driver<T>(
};
match message {
StoreMessage::Call(job) => {
active_calls.push(accessor.spawn(ConcurrentTask(job))?);
job.run_concurrent(accessor).await?;
}
StoreMessage::GuestCall(job) => {
active_calls.push(accessor.spawn(ConcurrentGuestTask(job))?);
active_calls.push(accessor.spawn(async move |accessor| job.run_concurrent(accessor).await)?);
}
StoreMessage::Shutdown {
job,
@@ -1110,7 +1316,7 @@ async fn run_driver<T>(
} => {
while active_calls.next().await.is_some() {}
loop_lifecycle.transition(DriverState::Stopping);
let result = job.run(accessor).await;
let result = job.run_concurrent(accessor).await;
poll_fn(|cx| accessor.poll_no_interesting_tasks(cx)).await;
drop(root_admission);
return result;
@@ -1136,11 +1342,10 @@ async fn run_driver<T>(
match message {
StoreMessage::Call(job) => {
active_calls.push(accessor.spawn(ConcurrentTask(job))?);
tokio::task::yield_now().await;
job.run_concurrent(accessor).await?;
}
StoreMessage::GuestCall(job) => {
active_calls.push(accessor.spawn(ConcurrentGuestTask(job))?);
active_calls.push(accessor.spawn(async move |accessor| job.run_concurrent(accessor).await)?);
}
StoreMessage::Shutdown {
job,
@@ -1150,7 +1355,7 @@ async fn run_driver<T>(
receiver.close();
while active_calls.next().await.is_some() {}
loop_lifecycle.transition(DriverState::Stopping);
let result = job.run(accessor).await;
let result = job.run_concurrent(accessor).await;
poll_fn(|cx| accessor.poll_no_interesting_tasks(cx)).await;
drop(root_admission);
return result;
@@ -1231,14 +1436,22 @@ mod tests {
};
use wasmtime::{Config, Engine, Store};
use super::{ControlMessage, DriverState, Lifecycle, run_driver};
use super::{ControlMessage, DriverState, Lifecycle, StoreMode, run_driver};
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn submission_close_waits_for_owner_control() {
for mode in [StoreMode::Sequential, StoreMode::Concurrent] {
submission_close_waits_for_owner_control_in(mode).await;
}
}
async fn submission_close_waits_for_owner_control_in(mode: StoreMode) {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
if mode == StoreMode::Concurrent {
config.wasm_component_model_async(true);
}
config.concurrency_support(mode == StoreMode::Concurrent);
let engine = Engine::new(&config).expect("test engine");
let store = Store::new(&engine, ());
let (sender, receiver) = mpsc::channel(1);
@@ -1252,6 +1465,7 @@ mod tests {
Arc::clone(&lifecycle),
ready_sender,
"test",
mode,
));
timeout(Duration::from_secs(10), async {

374
crates/pumpkin-plugin-runtime/src/lib.rs
View File

@@ -5,7 +5,7 @@ mod policy;
mod spawn;
pub use executor::{
LegacyGuestScope, LegacyStore, StoreDataMut, StoreExecutor, StoreFuture, StoreHandle,
LegacyGuestScope, LegacyStore, StoreDataMut, StoreExecutor, StoreFuture, StoreHandle, StoreMode,
};
pub use lifecycle::{DriverError, DriverJoin, DriverState};
pub use policy::{LegacySyncReentry, StorePolicy};
@@ -42,7 +42,7 @@ mod tests {
use super::{
DriverError, DriverState, LegacyStore, LegacySyncReentry, MAX_SYNC_REENTRY_DEPTH,
RuntimeSpawner, SpawnError, SpawnFuture,
RuntimeSpawner, SpawnError, SpawnFuture, StoreMode,
};
struct TestHostState;
@@ -123,25 +123,29 @@ mod tests {
}
#[derive(Clone)]
struct ConcurrentStore(Arc<LegacyStore<TestHostState>>);
struct TestStore(Arc<LegacyStore<TestHostState>>);
impl ConcurrentStore {
async fn new(store: Store<TestHostState>) -> Self {
Self::with_policy(store, LegacySyncReentry::new()).await
impl TestStore {
async fn new(store: Store<TestHostState>, mode: StoreMode) -> Self {
Self::with_policy(store, LegacySyncReentry::new(), mode).await
}
async fn with_policy(store: Store<TestHostState>, policy: LegacySyncReentry) -> Self {
async fn with_policy(
store: Store<TestHostState>,
policy: LegacySyncReentry,
mode: StoreMode,
) -> Self {
let spawner = Arc::new(TestSpawner {
runtime: tokio::runtime::Handle::current(),
});
let executor = LegacyStore::start(store, policy, spawner)
let executor = LegacyStore::start(store, policy, spawner, mode)
.await
.expect("start test Store driver");
Self(Arc::new(executor))
}
}
impl Deref for ConcurrentStore {
impl Deref for TestStore {
type Target = LegacyStore<TestHostState>;
fn deref(&self) -> &Self::Target {
@@ -149,36 +153,28 @@ mod tests {
}
}
struct BackgroundTask {
started: Arc<Notify>,
release: Arc<Notify>,
}
const MODES: [StoreMode; 2] = [StoreMode::Sequential, StoreMode::Concurrent];
impl wasmtime::component::AccessorTask<TestHostState> for BackgroundTask {
async fn run(
self,
_accessor: &wasmtime::component::Accessor<TestHostState>,
) -> wasmtime::Result<()> {
self.started.notify_one();
self.release.notified().await;
Ok(())
}
}
fn test_store() -> Store<TestHostState> {
fn test_engine(mode: StoreMode) -> Engine {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
Store::new(&engine, TestHostState)
if mode == StoreMode::Concurrent {
config.wasm_component_model_async(true);
}
config.concurrency_support(mode == StoreMode::Concurrent);
Engine::new(&config).expect("test engine")
}
fn test_store(mode: StoreMode) -> Store<TestHostState> {
Store::new(&test_engine(mode), TestHostState)
}
async fn legacy_store(
store: Store<TestHostState>,
policy: &LegacySyncReentry,
) -> ConcurrentStore {
ConcurrentStore::with_policy(store, policy.clone()).await
mode: StoreMode,
) -> TestStore {
TestStore::with_policy(store, policy.clone(), mode).await
}
fn assert_synchronous_reentry_depth_is_bounded_and_recovers() {
@@ -197,12 +193,12 @@ mod tests {
}
#[allow(clippy::too_many_lines)]
async fn root_admission_lifecycle_scenario() {
async fn root_admission_lifecycle_scenario(mode: StoreMode) {
let policy = LegacySyncReentry::new();
let holding_store = legacy_store(test_store(), &policy).await;
let first_store = legacy_store(test_store(), &policy).await;
let second_store = legacy_store(test_store(), &policy).await;
let third_store = legacy_store(test_store(), &policy).await;
let holding_store = legacy_store(test_store(mode), &policy, mode).await;
let first_store = legacy_store(test_store(mode), &policy, mode).await;
let second_store = legacy_store(test_store(mode), &policy, mode).await;
let third_store = legacy_store(test_store(mode), &policy, mode).await;
let holding_started = Arc::new(Notify::new());
let release_holding = Arc::new(Notify::new());
let holding_completed = Arc::new(Notify::new());
@@ -319,9 +315,14 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn legacy_root_admission_is_fifo_and_cancellation_safe_across_domains() {
timeout(Duration::from_secs(10), root_admission_lifecycle_scenario())
for mode in MODES {
timeout(
Duration::from_secs(10),
root_admission_lifecycle_scenario(mode),
)
.await
.expect("root admission lifecycle scenario timed out");
}
}
fn sync_reentry_component() -> Vec<u8> {
@@ -505,8 +506,8 @@ mod tests {
component.finish()
}
async fn shutdown_drains_accepted_calls_and_rejects_new_work() {
let store = ConcurrentStore::new(test_store()).await;
async fn shutdown_waits_for_in_flight_guest_call_and_rejects_new_work(mode: StoreMode) {
let store = TestStore::new(test_store(mode), mode).await;
let started = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
@@ -515,7 +516,7 @@ mod tests {
let call_release = Arc::clone(&release);
let active_call = tokio::spawn(async move {
call_store
.call(move |_| {
.call_guest(move |_| {
Box::pin(async move {
call_started.notify_one();
call_release.notified().await;
@@ -532,45 +533,47 @@ mod tests {
.shutdown(|_| Box::pin(async move { Ok("unloaded") }))
.await
});
while matches!(store.state(), DriverState::Accepting) {
// Shutdown queues for root admission behind the in-flight guest call.
for _ in 0..16 {
tokio::task::yield_now().await;
}
let rejected = store
.call(|_| Box::pin(async move { Ok(()) }))
.await
.expect_err("new work should be rejected during shutdown");
assert!(rejected.to_string().contains("shutting down"));
assert!(!shutdown.is_finished());
release.notify_one();
assert_eq!(active_call.await.expect("active call task").unwrap(), 7);
assert_eq!(shutdown.await.expect("shutdown task").unwrap(), "unloaded");
assert_eq!(store.state(), DriverState::Stopped);
let rejected = store
.call(|_| ())
.await
.expect_err("new work should be rejected after shutdown");
assert!(rejected.to_string().contains("shutting down"));
}
async fn shutdown_waits_for_joined_store_background_tasks() {
let store = ConcurrentStore::new(test_store()).await;
async fn shutdown_waits_for_joined_store_background_tasks(mode: StoreMode) {
let store = TestStore::new(test_store(mode), mode).await;
let started = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
let task_started = Arc::clone(&started);
let task_release = Arc::clone(&release);
let call_store = store.clone();
let active_call = tokio::spawn(async move {
call_store
.call(move |accessor| {
let spawned = accessor.spawn(BackgroundTask {
started: task_started,
release: task_release,
});
Box::pin(async move {
spawned?.await;
Ok(())
})
// The guest call returns at once and leaves a task running in the Store's event loop.
store
.call_guest(move |scope| {
let accessor = scope.accessor().expect("concurrent store scope");
let spawned = accessor.spawn(async move |_accessor| {
task_started.notify_one();
task_release.notified().await;
Ok(())
});
Box::pin(async move {
spawned?;
Ok(())
})
.await
});
})
.await
.expect("spawn background task");
started.notified().await;
let shutdown_store = store.clone();
@@ -579,29 +582,26 @@ mod tests {
.shutdown(|_| Box::pin(async move { Ok(()) }))
.await
});
while matches!(store.state(), DriverState::Accepting) {
for _ in 0..16 {
tokio::task::yield_now().await;
}
assert!(!shutdown.is_finished());
release.notify_one();
active_call
.await
.expect("active call task")
.expect("active call result");
shutdown
.await
.expect("shutdown task")
.expect("shutdown result");
}
async fn explicit_discard_drains_accepted_work() {
async fn explicit_discard_drains_accepted_work(mode: StoreMode) {
let spawner = Arc::new(TestSpawner {
runtime: tokio::runtime::Handle::current(),
});
let executor = LegacyStore::start(test_store(), LegacySyncReentry::new(), spawner)
.await
.expect("start test Store driver");
let executor =
LegacyStore::start(test_store(mode), LegacySyncReentry::new(), spawner, mode)
.await
.expect("start test Store driver");
let handle = executor.handle();
let started = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
@@ -611,7 +611,7 @@ mod tests {
let call_release = Arc::clone(&release);
let accepted = tokio::spawn(async move {
call_handle
.call(move |_| {
.call_guest(move |_| {
Box::pin(async move {
call_started.notify_one();
call_release.notified().await;
@@ -624,7 +624,7 @@ mod tests {
executor.discard();
let rejected = handle
.call(|_| Box::pin(async move { Ok(()) }))
.call(|_| ())
.await
.expect_err("discard should close admission");
assert!(rejected.to_string().contains("shutting down"));
@@ -639,13 +639,14 @@ mod tests {
assert_eq!(handle.state(), DriverState::Stopped);
}
async fn owner_drop_drains_accepted_work() {
async fn owner_drop_drains_accepted_work(mode: StoreMode) {
let spawner = Arc::new(TestSpawner {
runtime: tokio::runtime::Handle::current(),
});
let executor = LegacyStore::start(test_store(), LegacySyncReentry::new(), spawner)
.await
.expect("start test Store driver");
let executor =
LegacyStore::start(test_store(mode), LegacySyncReentry::new(), spawner, mode)
.await
.expect("start test Store driver");
let handle = executor.handle();
let started = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
@@ -655,7 +656,7 @@ mod tests {
let call_release = Arc::clone(&release);
let accepted = tokio::spawn(async move {
call_handle
.call(move |_| {
.call_guest(move |_| {
Box::pin(async move {
call_started.notify_one();
call_release.notified().await;
@@ -668,7 +669,7 @@ mod tests {
drop(executor);
let rejected = handle
.call(|_| Box::pin(async move { Ok(()) }))
.call(|_| ())
.await
.expect_err("owner drop should close admission");
assert!(rejected.to_string().contains("shutting down"));
@@ -687,11 +688,12 @@ mod tests {
);
let executor = LegacyStore::start(
test_store(),
test_store(mode),
LegacySyncReentry::new(),
Arc::new(TestSpawner {
runtime: tokio::runtime::Handle::current(),
}),
mode,
)
.await
.expect("start test Store driver");
@@ -710,15 +712,11 @@ mod tests {
/// Acceptance test for plain synchronous WIT recursion through the
/// call-scoped reentry pump.
async fn sync_lifted_same_instance_reentry_completes() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let driver_slot = Arc::new(OnceLock::<TestStore>::new());
let host_calls = Arc::new(AtomicUsize::new(0));
let mut linker = Linker::<TestHostState>::new(&engine);
@@ -765,7 +763,7 @@ mod tests {
.expect("run export");
assert!(run_slot.set(run).is_ok());
let driver = ConcurrentStore::new(store).await;
let driver = TestStore::new(store, StoreMode::Sequential).await;
assert!(driver_slot.set(driver.clone()).is_ok());
timeout(
@@ -792,15 +790,11 @@ mod tests {
/// Mirrors synchronous game methods that call `fire_blocking` from a
/// blocking worker while the host import keeps the active store pumpable.
async fn blocking_host_operation_propagates_the_reentry_chain() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let driver_slot = Arc::new(OnceLock::<TestStore>::new());
let host_calls = Arc::new(AtomicUsize::new(0));
let mut linker = Linker::<TestHostState>::new(&engine);
@@ -851,7 +845,7 @@ mod tests {
.expect("run export");
assert!(run_slot.set(run).is_ok());
let driver = ConcurrentStore::new(store).await;
let driver = TestStore::new(store, StoreMode::Sequential).await;
assert!(driver_slot.set(driver.clone()).is_ok());
timeout(
@@ -872,14 +866,10 @@ mod tests {
}
async fn unrelated_guest_call_waits_for_the_active_chain() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let driver_slot = Arc::new(OnceLock::<TestStore>::new());
let entered = Arc::new(Notify::new());
let release = Arc::new(Notify::new());
let unrelated_entered = Arc::new(Notify::new());
@@ -922,7 +912,7 @@ mod tests {
let run = instance
.get_typed_func::<(), ()>(&mut store, "run")
.expect("run export");
let driver = ConcurrentStore::new(store).await;
let driver = TestStore::new(store, StoreMode::Sequential).await;
assert!(driver_slot.set(driver.clone()).is_ok());
let outer_driver = driver.clone();
@@ -963,15 +953,11 @@ mod tests {
#[allow(clippy::too_many_lines)]
async fn shutdown_reentry_scenario() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let driver_slot = Arc::new(OnceLock::<TestStore>::new());
let entered = Arc::new(Notify::new());
let start_callback = Arc::new(Notify::new());
let callback_completed = Arc::new(Notify::new());
@@ -1027,7 +1013,7 @@ mod tests {
.get_typed_func::<(), ()>(&mut store, "run")
.expect("run export");
assert!(run_slot.set(run).is_ok());
let driver = ConcurrentStore::new(store).await;
let driver = TestStore::new(store, StoreMode::Sequential).await;
assert!(driver_slot.set(driver.clone()).is_ok());
let outer_driver = driver.clone();
@@ -1078,17 +1064,13 @@ mod tests {
#[allow(clippy::too_many_lines)]
async fn opposing_plugin_roots_scenario() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let a_run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let b_run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let a_driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let b_driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let a_driver_slot = Arc::new(OnceLock::<TestStore>::new());
let b_driver_slot = Arc::new(OnceLock::<TestStore>::new());
let a_calls = Arc::new(AtomicUsize::new(0));
let b_calls = Arc::new(AtomicUsize::new(0));
let trace = Arc::new(Mutex::new(Vec::new()));
@@ -1238,8 +1220,8 @@ mod tests {
assert!(b_run_slot.set(b_run).is_ok());
let policy = LegacySyncReentry::new();
let a_driver = legacy_store(a_store, &policy).await;
let b_driver = legacy_store(b_store, &policy).await;
let a_driver = legacy_store(a_store, &policy, StoreMode::Sequential).await;
let b_driver = legacy_store(b_store, &policy, StoreMode::Sequential).await;
assert!(a_driver_slot.set(a_driver.clone()).is_ok());
assert!(b_driver_slot.set(b_driver.clone()).is_ok());
@@ -1329,7 +1311,7 @@ mod tests {
}
fn nested_active_context_call(
driver: ConcurrentStore,
driver: TestStore,
remaining: usize,
blocking: bool,
) -> super::StoreFuture<'static, ()> {
@@ -1371,30 +1353,33 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn active_context_recursion_is_bounded_and_releases_admission() {
timeout(Duration::from_secs(10), async {
let mode = StoreMode::Concurrent;
for blocking in [false, true] {
let policy = LegacySyncReentry::new();
let driver = legacy_store(test_store(), &policy).await;
let unrelated = legacy_store(test_store(), &policy).await;
nested_active_context_call(driver.clone(), 2, blocking)
.await
.expect("bounded callbacks should complete");
let error = nested_active_context_call(
driver.clone(),
MAX_SYNC_REENTRY_DEPTH + 1,
blocking,
)
.await
.expect_err("recursive callbacks must reach the depth limit");
assert!(error.to_string().contains("maximum depth"));
unrelated
.call_guest(|_| Box::pin(async { Ok(()) }))
.await
.expect("depth failure must release root admission");
for store in [driver, unrelated] {
store
.shutdown(|_| Box::pin(async { Ok(()) }))
{
let policy = LegacySyncReentry::new();
let driver = legacy_store(test_store(mode), &policy, mode).await;
let unrelated = legacy_store(test_store(mode), &policy, mode).await;
nested_active_context_call(driver.clone(), 2, blocking)
.await
.expect("shut down store after depth failure");
.expect("bounded callbacks should complete");
let error = nested_active_context_call(
driver.clone(),
MAX_SYNC_REENTRY_DEPTH + 1,
blocking,
)
.await
.expect_err("recursive callbacks must reach the depth limit");
assert!(error.to_string().contains("maximum depth"));
unrelated
.call_guest(|_| Box::pin(async { Ok(()) }))
.await
.expect("depth failure must release root admission");
for store in [driver, unrelated] {
store
.shutdown(|_| Box::pin(async { Ok(()) }))
.await
.expect("shut down store after depth failure");
}
}
}
})
@@ -1402,24 +1387,59 @@ mod tests {
.expect("active-context recursion must not strand the Store driver");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn concurrent_store_rejects_pump_blocking_before_running_the_operation() {
let store = TestStore::new(test_store(StoreMode::Concurrent), StoreMode::Concurrent).await;
let ran = Arc::new(AtomicBool::new(false));
let nested = store.clone();
let flag = Arc::clone(&ran);
// Inside an active guest call, so a missing mode check would get as far as spawning.
let error = store
.call_guest(move |_| {
Box::pin(async move {
let mut pump_store = test_store(StoreMode::Concurrent);
nested
.pump_blocking(&mut pump_store, move || flag.store(true, Ordering::SeqCst))
.await
})
})
.await
.expect_err("a concurrent store cannot pump reentrant callbacks");
assert!(error.to_string().contains("sequential"), "{error}");
tokio::time::sleep(Duration::from_millis(100)).await;
assert!(
!ran.load(Ordering::SeqCst),
"the rejected operation must not run"
);
store
.shutdown(|_| Box::pin(async { Ok(()) }))
.await
.expect("shut down store");
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn shutdown_drains_accepted_work() {
timeout(Duration::from_secs(10), async {
shutdown_drains_accepted_calls_and_rejects_new_work().await;
shutdown_waits_for_joined_store_background_tasks().await;
explicit_discard_drains_accepted_work().await;
owner_drop_drains_accepted_work().await;
for mode in MODES {
shutdown_waits_for_in_flight_guest_call_and_rejects_new_work(mode).await;
explicit_discard_drains_accepted_work(mode).await;
owner_drop_drains_accepted_work(mode).await;
}
shutdown_waits_for_joined_store_background_tasks(StoreMode::Concurrent).await;
shutdown_reentry_scenario().await;
})
.await
.expect("Store shutdown scenarios timed out");
}
async fn startup_task_loss_is_reported() {
async fn startup_task_loss_is_reported(mode: StoreMode) {
let started = LegacyStore::start(
test_store(),
test_store(mode),
LegacySyncReentry::new(),
Arc::new(DroppingSpawner),
mode,
)
.await;
let Err(error) = started else {
@@ -1428,15 +1448,16 @@ mod tests {
assert!(error.to_string().contains("ended without reporting"));
}
async fn active_driver_task_loss_is_reported() {
async fn active_driver_task_loss_is_reported(mode: StoreMode) {
let driver_abort = Arc::new(Mutex::new(None));
let spawner = Arc::new(AbortRecordingSpawner {
runtime: tokio::runtime::Handle::current(),
driver_abort: Arc::clone(&driver_abort),
});
let executor = LegacyStore::start(test_store(), LegacySyncReentry::new(), spawner)
.await
.expect("start test Store driver");
let executor =
LegacyStore::start(test_store(mode), LegacySyncReentry::new(), spawner, mode)
.await
.expect("start test Store driver");
let handle = executor.handle();
let started = Arc::new(Notify::new());
let never_release = Arc::new(Notify::new());
@@ -1446,7 +1467,7 @@ mod tests {
let call_release = Arc::clone(&never_release);
let active = tokio::spawn(async move {
call_handle
.call(move |_| {
.call_guest(move |_| {
Box::pin(async move {
call_started.notify_one();
call_release.notified().await;
@@ -1481,8 +1502,8 @@ mod tests {
);
}
async fn driver_panic_is_reported() {
let store = ConcurrentStore::new(test_store()).await;
async fn driver_panic_is_reported(mode: StoreMode) {
let store = TestStore::new(test_store(mode), mode).await;
let handle = store.handle();
let call_error = timeout(
Duration::from_secs(2),
@@ -1503,7 +1524,7 @@ mod tests {
retained
);
let later_error = handle
.call(|_| Box::pin(async move { Ok(()) }))
.call(|_| ())
.await
.expect_err("later calls should retain the driver failure");
assert!(later_error.to_string().contains("ended without reporting"));
@@ -1511,14 +1532,10 @@ mod tests {
#[allow(clippy::too_many_lines)]
async fn caught_reentrant_guest_trap_stops_driver_and_queued_work() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
let engine = test_engine(StoreMode::Sequential);
let component = Component::new(&engine, sync_reentry_component()).expect("test component");
let run_slot = Arc::new(OnceLock::<TypedFunc<(), ()>>::new());
let driver_slot = Arc::new(OnceLock::<ConcurrentStore>::new());
let driver_slot = Arc::new(OnceLock::<TestStore>::new());
let host_calls = Arc::new(AtomicUsize::new(0));
let queued_call_ran = Arc::new(AtomicBool::new(false));
@@ -1571,7 +1588,7 @@ mod tests {
.get_typed_func::<(), ()>(&mut store, "run")
.expect("run export");
assert!(run_slot.set(run).is_ok());
let driver = ConcurrentStore::new(store).await;
let driver = TestStore::new(store, StoreMode::Sequential).await;
assert!(driver_slot.set(driver.clone()).is_ok());
let error = timeout(
@@ -1591,12 +1608,8 @@ mod tests {
}
#[allow(clippy::too_many_lines)]
async fn owned_resource_lifecycle_retires_store_on_trap() {
let mut config = Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let engine = Engine::new(&config).expect("test engine");
async fn owned_resource_lifecycle_retires_store_on_trap(mode: StoreMode) {
let engine = test_engine(mode);
let component =
Component::new(&engine, owned_resource_component()).expect("test component");
let resource_drops = Arc::new(AtomicUsize::new(0));
@@ -1648,6 +1661,7 @@ mod tests {
Arc::new(TestSpawner {
runtime: tokio::runtime::Handle::current(),
}),
mode,
)
.await
.expect("start test Store driver");
@@ -1751,7 +1765,7 @@ mod tests {
assert_eq!(driver.state(), DriverState::Failed(Arc::clone(&retained)));
let later_error = handle
.call(|_| Box::pin(async move { Ok(()) }))
.call(|_| ())
.await
.expect_err("later calls should retain the guest failure");
assert!(
@@ -1772,14 +1786,22 @@ mod tests {
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn terminal_failure_is_retained() {
startup_task_loss_is_reported().await;
active_driver_task_loss_is_reported().await;
driver_panic_is_reported().await;
caught_reentrant_guest_trap_stops_driver_and_queued_work().await;
owned_resource_lifecycle_retires_store_on_trap().await;
for mode in MODES {
terminal_failure_scenario(mode).await;
}
}
async fn terminal_failure_scenario(mode: StoreMode) {
startup_task_loss_is_reported(mode).await;
active_driver_task_loss_is_reported(mode).await;
driver_panic_is_reported(mode).await;
if mode == StoreMode::Sequential {
caught_reentrant_guest_trap_stops_driver_and_queued_work().await;
}
owned_resource_lifecycle_retires_store_on_trap(mode).await;
shutdown_does_not_hide_terminal_failure();
let store = ConcurrentStore::new(test_store()).await;
let store = TestStore::new(test_store(mode), mode).await;
let handle = store.handle();
let shutdown_error = store
.shutdown(|_| {
@@ -1800,7 +1822,7 @@ mod tests {
assert_eq!(joined, retained);
let call_error = handle
.call(|_| Box::pin(async move { Ok(()) }))
.call(|_| ())
.await
.expect_err("future calls should retain the driver failure");
assert!(call_error.to_string().contains("finalizer failed"));

5
crates/pumpkin-wasm-host-common/src/concurrent_store.rs
View File

@@ -5,7 +5,7 @@ use wasmtime::Store;
use super::state::PluginHostState;
pub use pumpkin_plugin_runtime::LegacySyncReentry;
pub use pumpkin_plugin_runtime::{LegacySyncReentry, StoreMode};
pub type LegacyStore = StoreExecutor<PluginHostState, LegacySyncReentry>;
@@ -36,6 +36,7 @@ pub async fn start_legacy_store(
store: Store<PluginHostState>,
policy: LegacySyncReentry,
spawner: Arc<dyn RuntimeSpawner>,
mode: StoreMode,
) -> wasmtime::Result<LegacyStore> {
StoreExecutor::start(store, policy, spawner).await
StoreExecutor::start(store, policy, spawner, mode).await
}

100
crates/pumpkin-wasm-host-common/src/plugin.rs
View File

@@ -13,6 +13,8 @@ pub enum PluginInitError {
UnknownApiVersion(),
#[error("Plugin is built against an unsupported version of the API: {0}")]
UnsupportedApiVersion(String),
#[error("Plugin API 0.1 supports WASI 0.2 only, but the plugin imports `{0}`")]
UnsupportedWasi(String),
#[error("Plugin was built against a different iteration of the API: {0}")]
ApiMismatch(wasmtime::Error),
#[error("Failed to read plugin file: {0}")]
@@ -35,6 +37,63 @@ pub enum PluginInitError {
CacheCreationFailed(wasmtime::Error),
}
/// What a plugin component declares at its top level, read without compiling it.
///
/// A compiled `Component` belongs to one `Engine`, and the engine depends on the plugin's API
/// version, so this has to be known first.
#[derive(Debug, Default, Eq, PartialEq)]
pub struct ComponentNames {
/// The version of the exported `pumpkin:plugin/metadata` interface, such as `0.1.0`.
pub metadata_version: Option<String>,
/// The first imported WASI 0.3 interface, if any.
pub wasi_p3_import: Option<String>,
}
/// Scans the top-level imports and exports of a component binary.
///
/// # Errors
/// Returns [`PluginInitError::ComponentNewFailed`] when the bytes are not a valid component.
pub fn scan_component_names(wasm_bytes: &[u8]) -> Result<ComponentNames, PluginInitError> {
let malformed = |error: wasmparser::BinaryReaderError| {
PluginInitError::ComponentNewFailed(wasmtime::Error::msg(error.to_string()))
};
let mut names = ComponentNames::default();
// Nested components and modules repeat the same section kinds; only depth 0 is the plugin.
let mut depth = 0usize;
for payload in wasmparser::Parser::new(0).parse_all(wasm_bytes) {
match payload.map_err(malformed)? {
wasmparser::Payload::ModuleSection { .. }
| wasmparser::Payload::ComponentSection { .. } => depth += 1,
wasmparser::Payload::End(_) => depth = depth.saturating_sub(1),
wasmparser::Payload::ComponentImportSection(imports) if depth == 0 => {
for import in imports {
let name = import.map_err(malformed)?.name.full_name();
if names.wasi_p3_import.is_none() && is_wasi_p3_import(&name) {
names.wasi_p3_import = Some(name.into_owned());
}
}
}
wasmparser::Payload::ComponentExportSection(exports) if depth == 0 => {
for export in exports {
let name = export.map_err(malformed)?.name.full_name();
if let Some(version) = name.strip_prefix("pumpkin:plugin/metadata@") {
names.metadata_version = Some(version.to_owned());
}
}
}
_ => {}
}
}
Ok(names)
}
fn is_wasi_p3_import(name: &str) -> bool {
name.starts_with("wasi:")
&& name
.rsplit_once('@')
.is_some_and(|(_, version)| version.starts_with("0.3."))
}
#[derive(Copy, Clone)]
pub enum PluginApiVersion {
V0_1,
@@ -67,3 +126,44 @@ impl Drop for WasmPlugin {
self.store.discard();
}
}
#[cfg(test)]
mod tests {
use wasm_encoder::{ComponentBuilder, ComponentTypeRef, InstanceType};
use super::{is_wasi_p3_import, scan_component_names};
fn component_importing(names: &[&str]) -> Vec<u8> {
let mut component = ComponentBuilder::default();
for name in names {
let ty = component.type_instance(None, &InstanceType::new());
component.import(*name, ComponentTypeRef::Instance(ty));
}
component.finish()
}
#[test]
fn wasi_p3_imports_are_recognised_by_version() {
assert!(is_wasi_p3_import("wasi:random/random@0.3.0"));
assert!(is_wasi_p3_import("wasi:cli/stdout@0.3.0-rc-2026-03-15"));
assert!(!is_wasi_p3_import("wasi:random/random@0.2.6"));
assert!(!is_wasi_p3_import("wasi:random/random"));
assert!(!is_wasi_p3_import("pumpkin:plugin/context@0.3.0"));
}
#[test]
fn scan_reports_the_first_wasi_p3_import() {
let bytes = component_importing(&["wasi:random/random@0.2.6", "wasi:random/random@0.3.0"]);
let names = scan_component_names(&bytes).expect("scan component");
assert_eq!(
names.wasi_p3_import.as_deref(),
Some("wasi:random/random@0.3.0")
);
assert_eq!(names.metadata_version, None);
}
#[test]
fn scan_rejects_bytes_that_are_not_a_component() {
assert!(scan_component_names(b"not wasm").is_err());
}
}

2
crates/pumpkin-wasm-host-v0_1/src/bindings.rs
View File

@@ -161,5 +161,5 @@ bindgen!({
"pumpkin:plugin/world@0.1.0.[method]living-entity.set-equipment": async | trappable,
default: trappable,
},
exports: { default: async | store | trappable },
exports: { default: async | trappable },
});

14
crates/pumpkin-wasm-host-v0_1/src/commands/executor.rs
View File

@@ -5,7 +5,9 @@ use pumpkin_util::text::{
color::{Color, NamedColor},
};
use crate::pumpkin::plugin::command::{CommandError as CommandErrorWit, SuggestionRequest};
use crate::pumpkin::plugin::command::{
CommandError as CommandErrorWit, CommandSender, CommandSuggestions, SuggestionRequest,
};
use pumpkin_core::{
command::{
context::command_context::CommandContext,
@@ -21,6 +23,7 @@ use pumpkin_core::{
use pumpkin_wasm_host_common::{
args::build_consumed_args_from_context, plugin::WasmPlugin, state::PluginHostState,
};
use wasmtime::component::Resource;
fn map_command_result(
state: &mut PluginHostState,
@@ -139,7 +142,7 @@ impl SuggestionProvider for WasmCommandSuggestionProvider {
let handler_id = self.handler_id;
let function = {
let plugin = self.plugin.instance::<crate::Plugin>();
plugin.func_handle_command_suggestion()
*plugin.func_handle_command_suggestion().func()
};
tokio::task::block_in_place(|| {
@@ -161,7 +164,12 @@ impl SuggestionProvider for WasmCommandSuggestionProvider {
Ok::<_, wasmtime::Error>((sender_resource, server_resource))
})?;
let response = guest
.call(
.call_func::<(
u32,
Resource<CommandSender>,
Resource<crate::pumpkin::plugin::server::Server>,
SuggestionRequest,
), (CommandSuggestions,)>(
function,
(handler_id, sender_resource, server_resource, request),
)

14
crates/pumpkin-wasm-host-v0_1/src/events/mod.rs
View File

@@ -276,7 +276,11 @@ impl WasmPluginEventHandler {
event: Box<dyn PendingWasmEvent>,
) -> wasmtime::Result<Box<dyn Any + Send>> {
let handler_id = self.handler_id;
let function = self.plugin.instance::<crate::Plugin>().func_handle_event();
let function = *self
.plugin
.instance::<crate::Plugin>()
.func_handle_event()
.func();
self.plugin
.store
.call_guest(move |mut guest| {
@@ -294,7 +298,13 @@ impl WasmPluginEventHandler {
// Lowering transfers these resources to the guest. Only a
// successfully returned event is owned by the host again.
let (returned_event,) = guest
.call(function, (handler_id, server_res, wasm_event))
.call_func::<(
u32,
Resource<crate::pumpkin::plugin::server::Server>,
WasmEvent,
), (WasmEvent,)>(
function, (handler_id, server_res, wasm_event)
)
.await?;
Ok(guest.with(|mut store| event.from_wasm(returned_event, store.data_mut())))
})

22
crates/pumpkin-wasm-host-v0_1/src/lib.rs
View File

@@ -386,24 +386,18 @@ pub async fn init_plugin(
.await
.map_err(PluginInitError::InstantiationFailed)?;
store
.run_concurrent(async |accessor| {
legacy_sync_reentry
.scope_bootstrap(plugin.call_init_plugin(accessor))
.await
})
legacy_sync_reentry
.scope_bootstrap(plugin.call_init_plugin(&mut store))
.await
.map_err(PluginInitError::CallInitPluginFailed)?
.map_err(PluginInitError::CallInitPluginFailed)?;
let metadata = store
.run_concurrent(async |accessor| {
legacy_sync_reentry
.scope_bootstrap(plugin.pumpkin_plugin_metadata().call_get_metadata(accessor))
.await
})
let metadata = legacy_sync_reentry
.scope_bootstrap(
plugin
.pumpkin_plugin_metadata()
.call_get_metadata(&mut store),
)
.await
.map_err(PluginInitError::CallGetMetadataFailed)?
.map_err(PluginInitError::CallGetMetadataFailed)?;
let metadata = PluginMetadata {

1
crates/pumpkin-wasm-host/Cargo.toml
View File

@@ -28,6 +28,7 @@ wasmtime-wasi-http = { workspace = true }
[dev-dependencies]
tempfile.workspace = true
wasm-encoder.workspace = true
tokio = { workspace = true, features = ["test-util"] }
[lints]

245
crates/pumpkin-wasm-host/src/runtime.rs
View File

@@ -7,8 +7,8 @@ use wasmtime_wasi::{FsPerms, WasiCtxBuilder, sockets::SocketAddrUse};
use pumpkin_core::plugin::{Context, PluginMetadata, cache::calculate_hash_for_bytes, permissions};
use pumpkin_plugin_runtime::RuntimeSpawner;
use pumpkin_wasm_host_common::{
concurrent_store::{self, LegacySyncReentry},
plugin::{PluginApiVersion, PluginInitError, WasmPlugin},
concurrent_store::{self, LegacySyncReentry, StoreMode},
plugin::{PluginApiVersion, PluginInitError, WasmPlugin, scan_component_names},
scheduler::TaskScheduler,
signature,
state::PluginHostState,
@@ -77,8 +77,14 @@ fn socket_policy_for_permissions(
}
}
/// Wasmtime only lets a host import call back into the guest of the store it is running on while
/// concurrency support is off, and only runs async components and WASI 0.3 while it is on. Each API
/// version gets the engine that suits its host imports: v0.1 imports call back into the guest,
/// v0.2 imports are concurrent.
pub struct PluginRuntime {
engine: Engine,
engine_v0_1: Engine,
#[cfg(feature = "v0_2")]
engine_v0_2: Engine,
cache_dir: std::path::PathBuf,
linker_v0_1: wasmtime::component::Linker<PluginHostState>,
#[cfg(feature = "v0_2")]
@@ -95,32 +101,25 @@ impl PluginRuntime {
task_scheduler: Arc<TaskScheduler>,
store_spawner: Arc<dyn RuntimeSpawner>,
) -> Result<Self, PluginInitError> {
let mut config = wasmtime::Config::new();
config.wasm_component_model(true);
config.wasm_component_model_async(true);
config.concurrency_support(true);
let mut path =
std::path::absolute(path.as_ref()).map_err(PluginInitError::PathResolutionFailed)?;
path.pop();
path.push("cache");
let mut cache_config = CacheConfig::new();
cache_config.with_directory(&path);
let cache = Cache::new(cache_config).map_err(PluginInitError::CacheCreationFailed)?;
config.cache(Some(cache));
config.gc_support(true);
config.wasm_gc(true);
config.wasm_exceptions(true);
config.wasm_function_references(true);
let engine = Engine::new(&config).map_err(PluginInitError::EngineCreationFailed)?;
let linker_v0_1 = setup_linker_v0_1(&engine).map_err(PluginInitError::LinkerSetupFailed)?;
let engine_v0_1 = new_engine(&path, StoreMode::Sequential)?;
#[cfg(feature = "v0_2")]
let linker_v0_2 = setup_linker_v0_2(&engine).map_err(PluginInitError::LinkerSetupFailed)?;
let engine_v0_2 = new_engine(&path, StoreMode::Concurrent)?;
let linker_v0_1 =
setup_linker_v0_1(&engine_v0_1).map_err(PluginInitError::LinkerSetupFailed)?;
#[cfg(feature = "v0_2")]
let linker_v0_2 =
setup_linker_v0_2(&engine_v0_2).map_err(PluginInitError::LinkerSetupFailed)?;
Ok(Self {
engine,
engine_v0_1,
#[cfg(feature = "v0_2")]
engine_v0_2,
cache_dir: path,
linker_v0_1,
#[cfg(feature = "v0_2")]
@@ -131,6 +130,16 @@ impl PluginRuntime {
})
}
const fn engine(&self, api_version: PluginApiVersion) -> &Engine {
match api_version {
PluginApiVersion::V0_1 => &self.engine_v0_1,
#[cfg(feature = "v0_2")]
PluginApiVersion::V0_2 => &self.engine_v0_2,
#[cfg(not(feature = "v0_2"))]
PluginApiVersion::V0_2 => &self.engine_v0_1,
}
}
pub async fn init_plugin<P: AsRef<Path>>(
&self,
path: P,
@@ -154,9 +163,9 @@ impl PluginRuntime {
let wasm_bytes = signature::strip_pumpkin_sections(&wasm_bytes).unwrap_or(wasm_bytes);
let component = load_component(&self.engine, &wasm_bytes, &self.cache_dir)?;
let api_version = plugin_api_version(&self.engine, &component)?;
let api_version = plugin_api_version(&wasm_bytes)?;
let engine = self.engine(api_version);
let component = load_component(engine, &wasm_bytes, &self.cache_dir)?;
let (plugin_instance, store, metadata) = match api_version {
PluginApiVersion::V0_1 => {
@@ -169,7 +178,7 @@ impl PluginRuntime {
.map_err(PluginInitError::ApiMismatch)?;
let (plugin, store, metadata) = pumpkin_wasm_host_v0_1::init_plugin(
&self.engine,
engine,
plugin_pre,
&self.legacy_sync_reentry,
)
@@ -191,7 +200,7 @@ impl PluginRuntime {
.map_err(PluginInitError::ApiMismatch)?;
let (plugin, store, metadata) = pumpkin_wasm_host_v0_2::init_plugin(
&self.engine,
engine,
plugin_pre,
&self.legacy_sync_reentry,
)
@@ -210,6 +219,10 @@ impl PluginRuntime {
store,
self.legacy_sync_reentry.clone(),
Arc::clone(&self.store_spawner),
match api_version {
PluginApiVersion::V0_1 => StoreMode::Sequential,
PluginApiVersion::V0_2 => StoreMode::Concurrent,
},
)
.await
.map_err(PluginInitError::InstantiationFailed)?;
@@ -222,14 +235,9 @@ impl PluginRuntime {
let weak_plugin = Arc::downgrade(&wasm_plugin);
wasm_plugin
.store
.call(move |accessor| {
Box::pin(async move {
accessor.with(|mut store| {
store.data_mut().plugin = Some(weak_plugin);
store.data_mut().marketplace_metadata = marketplace_metadata;
});
Ok(())
})
.call(move |state| {
state.plugin = Some(weak_plugin);
state.marketplace_metadata = marketplace_metadata;
})
.await
.map_err(PluginInitError::InstantiationFailed)?;
@@ -259,37 +267,57 @@ where
verify_signatures.then(|| verify(wasm_bytes, path_str))
}
fn plugin_api_version(
engine: &Engine,
component: &Component,
) -> Result<PluginApiVersion, PluginInitError> {
let ty = component.component_type();
for (name, _) in ty.exports(engine) {
if let Some(version) = name.strip_prefix("pumpkin:plugin/metadata@") {
// `starts_with` to ignore "minor" version differences
// WIT enforces semver. Will always be major.minor.patch
// For early/unstable release: 0.major.minor
if version.starts_with("0.1.") {
return Ok(PluginApiVersion::V0_1);
} else if version.starts_with("0.2.") {
return v0_2_api_version(version);
}
return Err(PluginInitError::UnsupportedApiVersion(version.to_string()));
/// Reads the plugin's API version from its bytes, before choosing the engine to compile it on.
fn plugin_api_version(wasm_bytes: &[u8]) -> Result<PluginApiVersion, PluginInitError> {
let names = scan_component_names(wasm_bytes)?;
let Some(version) = names.metadata_version else {
return Err(PluginInitError::UnknownApiVersion());
};
// `starts_with` to ignore "minor" version differences
// WIT enforces semver. Will always be major.minor.patch
// For early/unstable release: 0.major.minor
if version.starts_with("0.1.") {
// WASI 0.3 needs concurrency support, which the v0.1 engine doesn't have, so the linker
// would report this as an API mismatch.
if let Some(import) = names.wasi_p3_import {
return Err(PluginInitError::UnsupportedWasi(import));
}
Ok(PluginApiVersion::V0_1)
} else if version.starts_with("0.2.") {
v0_2_api_version(&version)
} else {
Err(PluginInitError::UnsupportedApiVersion(version))
}
}
Err(PluginInitError::UnknownApiVersion())
/// Builds the engine for one store mode, with the settings every plugin engine shares.
fn new_engine(cache_dir: &Path, mode: StoreMode) -> Result<Engine, PluginInitError> {
let mut config = wasmtime::Config::new();
config.wasm_component_model(true);
if mode == StoreMode::Concurrent {
config.wasm_component_model_async(true);
}
config.concurrency_support(mode == StoreMode::Concurrent);
let mut cache_config = CacheConfig::new();
cache_config.with_directory(cache_dir);
let cache = Cache::new(cache_config).map_err(PluginInitError::CacheCreationFailed)?;
config.cache(Some(cache));
config.gc_support(true);
config.wasm_gc(true);
config.wasm_exceptions(true);
config.wasm_function_references(true);
Engine::new(&config).map_err(PluginInitError::EngineCreationFailed)
}
fn setup_linker_v0_1(engine: &Engine) -> wasmtime::Result<Linker<PluginHostState>> {
let mut linker = Linker::<PluginHostState>::new(engine);
wasmtime_wasi::p2::add_to_linker_async(&mut linker)?;
wasmtime_wasi_http::p2::add_only_http_to_linker_async(&mut linker)?;
wasmtime_wasi::p3::add_to_linker(&mut linker)?;
wasmtime_wasi_http::p3::add_to_linker(&mut linker)?;
// WASI 0.3 is left out: its host functions are concurrent, and the v0.1 engine has
// concurrency support off.
pumpkin_wasm_host_v0_1::add_to_linker(&mut linker)?;
Ok(linker)
}
@@ -509,17 +537,7 @@ pub async fn on_unload(
) -> Result<Result<(), String>, wasmtime::Error> {
let loaded_plugin = wasm_plugin
.store
.call(|accessor| {
Box::pin(async move {
Ok(accessor.with(|mut store| {
store
.data_mut()
.plugin
.as_ref()
.and_then(std::sync::Weak::upgrade)
}))
})
})
.call(|state| state.plugin.as_ref().and_then(std::sync::Weak::upgrade))
.await?;
if let Some(plugin) = loaded_plugin {
@@ -530,25 +548,24 @@ pub async fn on_unload(
let api_version = wasm_plugin.api_version;
wasm_plugin
.store
.shutdown(move |accessor| {
.shutdown(move |mut guest| {
Box::pin(async move {
match api_version {
PluginApiVersion::V0_1 => {
let plugin =
guest_exports::<pumpkin_wasm_host_v0_1::Plugin>(&plugin_instance)?;
let (context_res, context_rep) = accessor.with(|mut store| {
let (context_res, context_rep) = guest.with(|mut store| {
let resource = store.data_mut().add(context)?;
let rep = resource.rep();
Ok::<_, wasmtime::Error>((resource, rep))
})?;
let result = plugin
.func_on_unload()
.call_concurrent(accessor, (context_res,))
let result = guest
.call(plugin.func_on_unload(), (context_res,))
.await
.map(|(result,)| result);
accessor.with(|mut store| {
guest.with(|mut store| {
let _ = store.data_mut().resource_table.delete::<Arc<Context>>(
wasmtime::component::Resource::new_own(context_rep),
);
@@ -560,19 +577,18 @@ pub async fn on_unload(
PluginApiVersion::V0_2 => {
let plugin =
guest_exports::<pumpkin_wasm_host_v0_2::Plugin>(&plugin_instance)?;
let (context_res, context_rep) = accessor.with(|mut store| {
let (context_res, context_rep) = guest.with(|mut store| {
let resource = store.data_mut().add(context)?;
let rep = resource.rep();
Ok::<_, wasmtime::Error>((resource, rep))
})?;
let result = plugin
.func_on_unload()
.call_concurrent(accessor, (context_res,))
let result = guest
.call(plugin.func_on_unload(), (context_res,))
.await
.map(|(result,)| result);
accessor.with(|mut store| {
guest.with(|mut store| {
let _ = store.data_mut().resource_table.delete::<Arc<Context>>(
wasmtime::component::Resource::new_own(context_rep),
);
@@ -597,13 +613,15 @@ pub async fn handle_ipc_message(
let message = message.to_owned();
let function = match wasm_plugin.api_version {
PluginApiVersion::V0_1 => {
guest_exports::<pumpkin_wasm_host_v0_1::Plugin>(&wasm_plugin.plugin_instance)?
*guest_exports::<pumpkin_wasm_host_v0_1::Plugin>(&wasm_plugin.plugin_instance)?
.func_handle_ipc_message()
.func()
}
#[cfg(feature = "v0_2")]
PluginApiVersion::V0_2 => {
guest_exports::<pumpkin_wasm_host_v0_2::Plugin>(&wasm_plugin.plugin_instance)?
*guest_exports::<pumpkin_wasm_host_v0_2::Plugin>(&wasm_plugin.plugin_instance)?
.func_handle_ipc_message()
.func()
}
#[cfg(not(feature = "v0_2"))]
PluginApiVersion::V0_2 => return Err(v0_2_disabled()),
@@ -614,7 +632,10 @@ pub async fn handle_ipc_message(
.call_guest(move |mut context| {
Box::pin(async move {
context
.call(function, (sender, message))
.call_func::<(String, Vec<u8>), (Result<Vec<u8>, String>,)>(
function,
(sender, message),
)
.await
.map(|(result,)| result)
})
@@ -628,10 +649,70 @@ mod tests {
use wasmtime_wasi::sockets::SocketAddrUse;
use super::{SocketPolicy, permissions, socket_policy_for_permissions};
use wasm_encoder::{ComponentBuilder, ComponentExportKind, ComponentTypeRef, InstanceType};
use super::{
PluginApiVersion, PluginInitError, SocketPolicy, permissions, plugin_api_version,
socket_policy_for_permissions,
};
const PORT: u16 = 25_565;
/// A component that imports each named instance and, when `metadata` is set, exports
/// `pumpkin:plugin/metadata@<metadata>`.
fn component(metadata: Option<&str>, imports: &[&str]) -> Vec<u8> {
let mut component = ComponentBuilder::default();
let mut first = None;
for name in imports {
let ty = component.type_instance(None, &InstanceType::new());
let index = component.import(*name, ComponentTypeRef::Instance(ty));
first.get_or_insert(index);
}
if let Some(version) = metadata {
let instance = first.unwrap_or_else(|| {
let ty = component.type_instance(None, &InstanceType::new());
component.import("unused:dep/dep@0.0.1", ComponentTypeRef::Instance(ty))
});
component.export(
format!("pumpkin:plugin/metadata@{version}"),
ComponentExportKind::Instance,
instance,
None,
);
}
component.finish()
}
#[test]
fn v0_1_plugin_is_identified_by_its_metadata_export() {
let bytes = component(Some("0.1.0"), &["wasi:random/random@0.2.6"]);
assert!(matches!(
plugin_api_version(&bytes),
Ok(PluginApiVersion::V0_1)
));
}
#[test]
fn v0_1_plugin_importing_wasi_p3_is_rejected_with_the_import_name() {
let bytes = component(Some("0.1.0"), &["wasi:random/random@0.3.0"]);
assert!(matches!(
plugin_api_version(&bytes),
Err(PluginInitError::UnsupportedWasi(name)) if name == "wasi:random/random@0.3.0"
));
}
#[test]
fn components_without_a_usable_metadata_export_are_rejected() {
assert!(matches!(
plugin_api_version(&component(None, &["wasi:random/random@0.2.6"])),
Err(PluginInitError::UnknownApiVersion())
));
assert!(matches!(
plugin_api_version(&component(Some("9.0.0"), &[])),
Err(PluginInitError::UnsupportedApiVersion(version)) if version == "9.0.0"
));
}
fn addr(ip: IpAddr) -> SocketAddr {
SocketAddr::new(ip, PORT)
}

8
flake.nix
View File

@@ -47,13 +47,7 @@
src = lib.cleanSource ./.;
cargoLock = {
lockFile = ./Cargo.lock;
outputHashes = {
"cranelift-assembler-x64-0.136.0-dev" =
"sha256-TZkmQ4+wWzb9x8UukZYQs1j05llI8ZmuMyHFXaDwcL0=";
};
};
cargoLock.lockFile = ./Cargo.lock;
nativeBuildInputs = [
pkgs.rustfmt