Fix duplicate completion registration in generated progress promises

bindings/js/src/com_completion_test_hooks.rs

@@ -652,15 +652,30 @@ unsafe extern "system" fn result_u32(this: *mut c_void, output: *mut u32) -> HRE
 }
 
 macro_rules! stub {
-  ($name:ident($($arg:ident: $typ:ty),*)) => {
+  ($name:ident($($arg:ident: $typ:ty),* $(,)?)) => {
     unsafe extern "system" fn $name(_this: *mut c_void, $($arg: $typ),*) -> HRESULT {
       E_NOTIMPL
     }
   };
 }
-stub!(audio_initialize(_mode: AUDCLNT_SHAREMODE, _flags: u32, _duration: i64, _period: i64, _format: *const WAVEFORMATEX, _session: *const GUID));
+stub!(
+  audio_initialize(
+    _mode: AUDCLNT_SHAREMODE,
+    _flags: u32,
+    _duration: i64,
+    _period: i64,
+    _format: *const WAVEFORMATEX,
+    _session: *const GUID,
+  )
+);
 stub!(audio_i64(_value: *mut i64));
-stub!(audio_format_supported(_mode: AUDCLNT_SHAREMODE, _format: *const WAVEFORMATEX, _closest: *mut *mut WAVEFORMATEX));
+stub!(
+  audio_format_supported(
+    _mode: AUDCLNT_SHAREMODE,
+    _format: *const WAVEFORMATEX,
+    _closest: *mut *mut WAVEFORMATEX,
+  )
+);
 stub!(audio_mix_format(_format: *mut *mut WAVEFORMATEX));
 stub!(audio_period(_default: *mut i64, _minimum: *mut i64));
 stub!(audio_noargs());

tests/e2e/e2e_specs.json

@@ -518,11 +518,22 @@
       "instantiate": { "kind": "none" },
       "checks": [
         { "kind": "async_memory_roundtrip", "member": "roundtrip", "write_value": 42 },
+        { "kind": "async_progress_promise", "member": "write_async", "langs": ["ts"] },
         { "kind": "data_stream_scalar_roundtrip", "member": "roundtrip", "langs": ["py"] },
         { "kind": "data_stream_constructor_roundtrip", "member": "__init__", "langs": ["py"] },
         { "kind": "can_cast_non_object", "member": "_dynwinrt_can_cast", "langs": ["py"] }
       ]
     },
+    {
+      "id": "async_progress_http_cancellation",
+      "namespace": "Windows.Web.Http",
+      "class": "HttpClient",
+      "langs": ["ts"],
+      "instantiate": { "kind": "none" },
+      "checks": [
+        { "kind": "async_progress_cancel_error", "member": "get_string_async" }
+      ]
+    },
     {
       "id": "nested_struct_runtime",
       "namespace": "Windows.Graphics.DirectX.Direct3D11",

tests/e2e/e2e_specs.schema.json

@@ -77,6 +77,8 @@
              "constructor_raises_type_error",
              "cross_class_chain",
              "async_memory_roundtrip",
+             "async_progress_promise",
+             "async_progress_cancel_error",
              "async_cancellation",
              "property_set_equals",
              "vector_view_access",

tests/e2e/runners/ts_progress_promise_child.mjs

@@ -0,0 +1,195 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+import assert from 'node:assert/strict'
+import { createServer } from 'node:http'
+import { createRequire } from 'node:module'
+import { resolve } from 'node:path'
+import { setImmediate } from 'node:timers/promises'
+
+const [name, generatedDir, runtimePath] = process.argv.slice(2)
+if (!name || !generatedDir || !runtimePath) {
+  throw new Error('Usage: ts_progress_promise_child.mjs <name> <generated> <runtime>')
+}
+const require = createRequire(import.meta.url)
+const g = require(resolve(generatedDir, 'index.js'))
+const runtime = require(resolve(runtimePath))
+runtime.roInitialize(1)
+
+async function streamPromises() {
+  const owned = []
+  const own = (value) => {
+    owned.push(value)
+    return value
+  }
+  const stream = own(new g.InMemoryRandomAccessStream())
+  const bytes = Buffer.alloc(4 * 1024 * 1024, 0x5a)
+  const buffer = own(g.Buffer.fromBuffer(bytes))
+  try {
+    const progress = []
+    const op = stream.writeAsync(buffer)
+    const promise = op.progress((value) => progress.push(value)).toPromise()
+    assert.deepEqual(
+      await Promise.all([op, promise, op.toPromise(), op.then((value) => value)]),
+      Array(4).fill(bytes.length),
+    )
+    assert.equal(promise, op)
+    assert.equal(op.toPromise(), promise)
+    op.cancel()
+    assert.equal(await op.toPromise(), bytes.length)
+    await setImmediate()
+    // Stock memory writes can finish before progress registration.
+    assert.ok(progress.every((value) => Number.isInteger(value) && value >= 0 && value <= bytes.length))
+
+    stream.seek(0n)
+    const destination = own(g.Buffer.fromBuffer(Buffer.alloc(bytes.length)))
+    const read = stream.readAsync(destination, bytes.length, g.InputStreamOptions.None)
+    const readPromise = read.toPromise()
+    const [result, sameResult] = await Promise.all([read, readPromise, read.toPromise()])
+    own(result)
+    assert.equal(result, sameResult, 'convert the native buffer into one projected wrapper')
+    assert.equal(readPromise, read)
+    assert.equal(await read.toPromise(), result)
+    assert.deepEqual(result.toBuffer(), bytes)
+
+    stream.seek(0n)
+    assert.equal(await stream.writeAsync(buffer), bytes.length, 'ordinary await remains supported')
+    stream.close()
+    assert.throws(() => stream.writeAsync(buffer), /0x80000013|closed/i)
+    const aborted = new AbortController()
+    const reason = new Error('already aborted')
+    aborted.abort(reason)
+    const preAborted = stream.writeAsync(buffer, aborted.signal)
+    const rejected = preAborted.progress(() => assert.fail('unexpected progress')).toPromise()
+    assert.equal(rejected, preAborted)
+    await assert.rejects(rejected, (error) => error === reason)
+    await assert.rejects(preAborted.toPromise(), (error) => error === reason)
+    preAborted.cancel()
+  } finally {
+    for (const value of owned.reverse()) g.releaseProjected(value)
+  }
+  console.log(`progress-stream-ok arch=${process.arch} bytes=${bytes.length}`)
+}
+
+async function httpCancellation() {
+  const waiting = new Map()
+  const payload = 'dynwinrt-progress-'.repeat(16_384)
+  const server = createServer((request, response) => {
+    if (request.url === '/error') {
+      response.writeHead(404, { 'Content-Length': 0, Connection: 'close' })
+      response.end()
+    } else if (request.url === '/progress') {
+      response.writeHead(200, { 'Content-Length': Buffer.byteLength(payload), Connection: 'close' })
+      let offset = 0
+      const send = () => {
+        if (response.destroyed) return
+        if (offset === payload.length) {
+          response.end()
+          return
+        }
+        const end = Math.min(offset + 8192, payload.length)
+        response.write(payload.slice(offset, end))
+        offset = end
+        setTimeout(send, 5)
+      }
+      send()
+    } else {
+      const accepted = waiting.get(request.url)
+      assert.ok(accepted, `unexpected localhost request ${request.url}`)
+      accepted()
+      // Hold the response open so cancellation cannot race a successful completion.
+    }
+  })
+  await new Promise((resolve, reject) => {
+    server.once('error', reject)
+    server.listen(0, '127.0.0.1', resolve)
+  })
+  const client = new g.HttpClient()
+  const uris = []
+  const uri = (path) => {
+    const value = new g.Uri(`http://127.0.0.1:${server.address().port}${path}`)
+    uris.push(value)
+    return value
+  }
+  try {
+    const progress = []
+    const op = client.getStringAsync(uri('/progress')).progress((value) => progress.push(value))
+    const promise = op.toPromise()
+    assert.deepEqual(await Promise.all([op, promise, op.toPromise()]), Array(3).fill(payload))
+    assert.equal(promise, op)
+    await setImmediate()
+    assert.ok(
+      progress.some(
+        (value) => value.bytesReceived > 0n && value.totalBytesToReceive === BigInt(Buffer.byteLength(payload)),
+      ),
+    )
+    assert.ok(progress.every((value) => typeof value.stage === 'number' && typeof value.retries === 'number'))
+    op.cancel()
+    assert.equal(await op.toPromise(), payload)
+
+    const headers = client.getAsync(uri('/progress'), g.HttpCompletionOption.ResponseHeadersRead)
+    const [response, sameResponse] = await Promise.all([headers, headers.toPromise(), headers.toPromise()])
+    assert.equal(response, sameResponse, 'overload consumers share one projected response')
+    try {
+      const content = response.content
+      try {
+        const body = content.readAsStringAsync()
+        const bodyPromise = body.progress((value) => assert.equal(typeof value, 'bigint')).toPromise()
+        assert.equal(bodyPromise, body)
+        assert.deepEqual(await Promise.all([body, bodyPromise, body.toPromise()]), Array(3).fill(payload))
+        await setImmediate()
+      } finally {
+        g.releaseProjected(content)
+      }
+    } finally {
+      response.close()
+      g.releaseProjected(response)
+    }
+
+    const download = client.getBufferAsync(uri('/progress'))
+    const [buffer, sameBuffer] = await Promise.all([download, download.toPromise(), download.toPromise()])
+    try {
+      assert.equal(buffer, sameBuffer, 'buffer consumers share one projected result')
+      assert.equal(buffer.toBuffer().toString('utf8'), payload)
+    } finally {
+      g.releaseProjected(buffer)
+    }
+
+    for (const mode of ['cancel', 'abort']) {
+      const path = `/${mode}`
+      const accepted = new Promise((resolve) => waiting.set(path, resolve))
+      const controller = new AbortController()
+      const reason = new Error('aborted during native request')
+      const op = client.getStringAsync(uri(path), controller.signal)
+      const consumers = Promise.allSettled([op, op.toPromise(), op.toPromise()])
+      await accepted
+      if (mode === 'cancel') op.cancel()
+      else controller.abort(reason)
+      const results = await consumers
+      assert.ok(results.every((result) => result.status === 'rejected'))
+      assert.equal(results[0].reason, results[1].reason)
+      assert.equal(results[1].reason, results[2].reason)
+      if (mode === 'abort') assert.equal(results[0].reason, reason)
+      // GetResults on a canceled stock operation can also report E_ILLEGAL_METHOD_CALL.
+      else assert.match(results[0].reason.message, /0x80004004|0x800704c7|0x8000000e/i)
+    }
+
+    const failed = client.getStringAsync(uri('/error'))
+    const errors = await Promise.allSettled([failed, failed.toPromise(), failed.toPromise()])
+    assert.ok(errors.every((result) => result.status === 'rejected'))
+    assert.equal(errors[0].reason, errors[1].reason)
+    assert.equal(errors[1].reason, errors[2].reason)
+    assert.match(errors[0].reason.message, /0x80190194|404/i)
+  } finally {
+    client.close()
+    g.releaseProjected(client)
+    for (const value of uris.reverse()) g.releaseProjected(value)
+    server.closeAllConnections()
+    await new Promise((resolve, reject) => server.close((error) => (error ? reject(error) : resolve())))
+  }
+  console.log(`progress-cancel-error-ok arch=${process.arch}`)
+}
+
+if (name === 'async_progress_promise') await streamPromises()
+else if (name === 'async_progress_cancel_error') await httpCancellation()
+else throw new Error(`Unknown progress Promise regression: ${name}`)

tests/e2e/runners/ts_runner.ts

@@ -231,7 +231,12 @@ async function runIssueRegression(
   timedOut: boolean;
 }> {
   const childPath = fileURLToPath(
-    new URL("./ts_issue_regression_child.mjs", import.meta.url),
+    new URL(
+      name === "async_progress_promise" || name === "async_progress_cancel_error"
+        ? "./ts_progress_promise_child.mjs"
+        : "./ts_issue_regression_child.mjs",
+      import.meta.url,
+    ),
   );
   const child = spawn(
     process.execPath,
@@ -959,7 +964,9 @@ async function runCheck(
       if (chainOk) cr.pass = true;
     } else if (
       kind === "device_information_async_collection" ||
-      kind === "bitmap_encoder_async_create"
+      kind === "bitmap_encoder_async_create" ||
+      kind === "async_progress_promise" ||
+      kind === "async_progress_cancel_error"
     ) {
       const child = await runIssueRegression(kind, generatedDir, runtimePath);
       if (child.timedOut) {

tools/dynwinrt-codegen/npm/README.md

@@ -73,7 +73,8 @@ For each WinRT class, the codegen emits:
 - **A JavaScript-backed `IElementFactory.create()` helper** for WinUI
   ItemsRepeater realization and recycling
 - **Promise-based async operations**, with `.progress(cb)` on operations that
-  expose WinRT progress
+  expose WinRT progress. Their `.toPromise()` returns the same projected Promise
+  on every call, equivalent to awaiting the operation directly
 - **Generic collections** (`IVector<T>`, `IMap<K,V>`, `IIterable<T>`)
 - **Creatable observable vectors** that expose both `IObservableVector<T>`
   events and `IVector<T>` mutation helpers

tools/dynwinrt-codegen/src/codegen/winrt/javascript/render/declarations.rs

@@ -48,7 +48,7 @@ pub fn render(file: &ProjectedFile) -> String {
             "    /** Register a callback for progress updates. Returns `this` for chaining. */\n",
         );
         body.push_str("    progress(cb: (value: P) => void): WinRTAsyncWithProgress<T, P>;\n");
-        body.push_str("    /** Get the underlying Promise (equivalent to awaiting directly). */\n");
+        body.push_str("    /** Get the same underlying Promise on every call (equivalent to awaiting directly). */\n");
         body.push_str("    toPromise(): Promise<T>;\n");
         body.push_str("    /** Cancel the underlying WinRT async operation. */\n");
         body.push_str("    cancel(): void;\n");

tools/dynwinrt-codegen/src/codegen/winrt/javascript/render/javascript/helpers.rs

@@ -47,7 +47,7 @@ pub(super) fn emit_with_progress_body(
     out.push_str("            _rej.catch(() => {});\n");
     out.push_str("            return Object.assign(_rej, {\n");
     out.push_str("                progress(_cb) { return this; },\n");
-    out.push_str("                toPromise() { const _p = Promise.reject(signal.reason); _p.catch(() => {}); return _p; },\n");
+    out.push_str("                toPromise() { return _rej; },\n");
     out.push_str("                cancel() {},\n");
     out.push_str("            });\n");
     out.push_str("        }\n");
@@ -74,7 +74,7 @@ pub(super) fn emit_with_progress_body(
     } else {
         out.push_str("            progress(cb) { _op.onProgress(cb); return this; },\n");
     }
-    out.push_str("            toPromise() { return _wrap(_op.toPromise()); },\n");
+    out.push_str("            toPromise() { return _promise; },\n");
     out.push_str("            cancel() { try { _op.cancel(); } catch (_ce) { /* cancel after completion is a no-op per WinRT spec */ } },\n");
     out.push_str("        });\n");
 }