fix(tsconnect): address bridge review findings

Co-Authored-By: gpt-5.6-sol <noreply@openai.com>

cmd/tsconnect/wasm/bridge_test.go

@@ -0,0 +1,326 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+//go:build js && wasm
+
+package main
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http/httptest"
+	"strings"
+	"syscall/js"
+	"testing"
+	"time"
+
+	"tailscale.com/ipn"
+	"tailscale.com/tailcfg"
+)
+
+func TestMakePromiseRejectsJSError(t *testing.T) {
+	errValue := js.Global().Get("Error").New("boom")
+	promise := makePromise(func() (any, error) {
+		panic(js.Error{Value: errValue})
+	})
+	rejection := awaitPromise(t, promise)
+	if rejection.Type() != js.TypeObject || rejection.Get("message").String() != "boom" {
+		t.Fatalf("rejection = %v, want Error(boom)", rejection)
+	}
+}
+
+func TestDriveHandlerThrowReturns500(t *testing.T) {
+	fn := evalJS(t, `(function() { throw new Error("boom") })`)
+	fs := new(jsFileSystemForRemote)
+	fs.setHandler(fn)
+	recorder := httptest.NewRecorder()
+	fs.ServeHTTPWithPerms(nil, recorder, httptest.NewRequest("GET", "/", nil))
+	if recorder.Code != 500 {
+		t.Fatalf("status = %d, want 500", recorder.Code)
+	}
+}
+
+func TestDriveCancellationDisablesCallbacks(t *testing.T) {
+	type handlerArgs struct {
+		req js.Value
+		res js.Value
+	}
+	started := make(chan handlerArgs, 1)
+	deferred := evalJS(t, `(() => {
+		let resolve
+		const promise = new Promise(r => { resolve = r })
+		return {promise, resolve}
+	})()`)
+	handler := js.FuncOf(func(this js.Value, args []js.Value) any {
+		started <- handlerArgs{req: args[0], res: args[1]}
+		return deferred.Get("promise")
+	})
+	defer handler.Release()
+	fs := new(jsFileSystemForRemote)
+	fs.setHandler(handler.Value)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	recorder := httptest.NewRecorder()
+	done := make(chan struct{})
+	go func() {
+		fs.ServeHTTPWithPerms(nil, recorder, httptest.NewRequest("GET", "/", nil).WithContext(ctx))
+		close(done)
+	}()
+
+	var jsArgs handlerArgs
+	select {
+	case jsArgs = <-started:
+	case <-time.After(time.Second):
+		t.Fatal("drive handler did not start")
+	}
+	cancel()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("drive handler did not stop after cancellation")
+	}
+
+	readResult := jsArgs.req.Call("readBodyChunk")
+	if readResult.Type() != js.TypeObject {
+		t.Fatalf("readBodyChunk after cancellation returned %s, want Promise", readResult.Type())
+	}
+	if result := awaitFulfilledPromise(t, readResult); !result.IsNull() {
+		t.Fatalf("readBodyChunk after cancellation = %v, want null", result)
+	}
+	lateChunk := js.Global().Get("Uint8Array").New(1)
+	js.CopyBytesToJS(lateChunk, []byte("x"))
+	jsArgs.res.Call("write", lateChunk)
+	if recorder.Body.Len() != 0 {
+		t.Fatalf("late JS write changed response body to %q", recorder.Body.String())
+	}
+	deferred.Call("resolve")
+
+	writeStarted := make(chan struct{})
+	allowWrite := make(chan struct{})
+	response := &driveResponse{
+		w:      recorder,
+		doneCh: make(chan error, 1),
+		live:   true,
+		testBeforeWriteCheck: func() {
+			close(writeStarted)
+			<-allowWrite
+		},
+	}
+	chunk := js.Global().Get("Uint8Array").New(1)
+	js.CopyBytesToJS(chunk, []byte("x"))
+	writeReturned := make(chan struct{})
+	go func() {
+		response.write([]js.Value{chunk})
+		close(writeReturned)
+	}()
+	select {
+	case <-writeStarted:
+	case <-time.After(time.Second):
+		t.Fatal("response write did not start")
+	}
+	response.finish(context.Canceled, context.Canceled)
+	close(allowWrite)
+	select {
+	case <-writeReturned:
+	case <-time.After(time.Second):
+		t.Fatal("response callback did not finish")
+	}
+	if recorder.Body.Len() != 0 {
+		t.Fatalf("late write changed response body to %q", recorder.Body.String())
+	}
+}
+
+func TestDriveHandlerFulfillmentCompletesRequest(t *testing.T) {
+	handler := evalJS(t, `(function() { return Promise.resolve() })`)
+	fs := new(jsFileSystemForRemote)
+	fs.setHandler(handler)
+	recorder := httptest.NewRecorder()
+	done := make(chan struct{})
+	go func() {
+		fs.ServeHTTPWithPerms(nil, recorder, httptest.NewRequest("GET", "/", nil))
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("fulfilled drive handler did not complete request")
+	}
+}
+
+func TestJSStreamReaderErrorsAreSticky(t *testing.T) {
+	for _, test := range []struct {
+		name   string
+		reader string
+	}{
+		{"rejected", `({read() { return Promise.reject("boom") }, cancel() { return Promise.resolve() }})`},
+		{"throwing", `({read() { throw new Error("boom") }, cancel() { return Promise.resolve() }})`},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			r := &jsStreamReader{reader: evalJS(t, test.reader)}
+			_, first := r.Read(make([]byte, 1))
+			_, second := r.Read(make([]byte, 1))
+			if first == nil || second == nil || first.Error() != second.Error() || !strings.Contains(first.Error(), "boom") {
+				t.Fatalf("read errors = %v, %v; want matching boom errors", first, second)
+			}
+		})
+	}
+}
+
+func TestJSStreamReaderCloseErrors(t *testing.T) {
+	for _, test := range []struct {
+		name   string
+		reader string
+	}{
+		{"rejected", `({cancel() { return Promise.reject("boom") }})`},
+		{"throwing", `({cancel() { throw new Error("boom") }})`},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			r := &jsStreamReader{reader: evalJS(t, test.reader)}
+			if err := r.Close(); err == nil || !strings.Contains(err.Error(), "boom") {
+				t.Fatalf("Close error = %v, want boom", err)
+			}
+		})
+	}
+}
+
+func TestJSStreamReaderCloseTimesOut(t *testing.T) {
+	r := &jsStreamReader{
+		reader:            evalJS(t, `({cancel() { return new Promise(() => {}) }})`),
+		testCancelTimeout: 10 * time.Millisecond,
+	}
+	started := time.Now()
+	if err := r.Close(); err == nil || !strings.Contains(err.Error(), "timed out") {
+		t.Fatalf("Close error = %v, want timeout", err)
+	}
+	if elapsed := time.Since(started); elapsed >= 250*time.Millisecond {
+		t.Fatalf("Close took %v, want less than 250ms", elapsed)
+	}
+	if _, err := r.Read(make([]byte, 1)); err == nil || !strings.Contains(err.Error(), "timed out") {
+		t.Fatalf("Read error after Close timeout = %v, want sticky timeout", err)
+	}
+}
+
+func TestJSStreamReaderIgnoresDuplicateSettlement(t *testing.T) {
+	reader := evalJS(t, `({
+		read() {
+			return {then(resolve, reject) { resolve({done: true}); reject("late") }}
+		},
+		cancel() {
+			return {then(resolve, reject) { resolve(); reject("late") }}
+		},
+	})`)
+	r := &jsStreamReader{reader: reader}
+	if _, err := r.Read(make([]byte, 1)); !errors.Is(err, io.EOF) {
+		t.Fatalf("Read error = %v, want EOF", err)
+	}
+	if err := r.Close(); err != nil {
+		t.Fatalf("Close error = %v, want nil", err)
+	}
+}
+
+func TestNotifyRefreshesNetMap(t *testing.T) {
+	online := true
+	lastSeen := time.Now()
+	for _, test := range []struct {
+		name string
+		n    ipn.Notify
+		want bool
+	}{
+		{"self", ipn.Notify{SelfChange: new(tailcfg.Node)}, true},
+		{"peer changed", ipn.Notify{PeersChanged: []*tailcfg.Node{{}}}, true},
+		{"peer removed", ipn.Notify{PeersRemoved: []tailcfg.NodeID{1}}, true},
+		{"online patch", ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{Online: &online}}}, true},
+		{"last seen patch", ipn.Notify{PeerChangedPatch: []*tailcfg.PeerChange{{LastSeen: &lastSeen}}}, false},
+		{"state only", ipn.Notify{State: new(ipn.State)}, false},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			if got := notifyRefreshesNetMap(test.n); got != test.want {
+				t.Fatalf("notifyRefreshesNetMap() = %v, want %v", got, test.want)
+			}
+		})
+	}
+}
+
+func TestNotifyWatchMask(t *testing.T) {
+	want := ipn.NotifyInitialState | ipn.NotifyInitialPrefs | ipn.NotifyPeerChanges | ipn.NotifyPeerPatches
+	if jsIPNNotifyWatchMask != want {
+		t.Fatalf("jsIPNNotifyWatchMask = %v, want %v", jsIPNNotifyWatchMask, want)
+	}
+	if jsIPNNotifyWatchMask&ipn.NotifyInProcessNoDisconnect != 0 {
+		t.Fatal("jsIPNNotifyWatchMask must allow a lagging watcher to disconnect")
+	}
+}
+
+func evalJS(t *testing.T, source string) js.Value {
+	t.Helper()
+	value, err := callJSFunction(js.Global().Get("eval"), source)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return value
+}
+
+func awaitPromise(t *testing.T, promise js.Value) js.Value {
+	t.Helper()
+	type result struct {
+		value    js.Value
+		rejected bool
+	}
+	ch := make(chan result, 1)
+	resolve := js.FuncOf(func(this js.Value, args []js.Value) any {
+		ch <- result{value: args[0]}
+		return nil
+	})
+	reject := js.FuncOf(func(this js.Value, args []js.Value) any {
+		ch <- result{value: args[0], rejected: true}
+		return nil
+	})
+	defer resolve.Release()
+	defer reject.Release()
+	if _, err := callJSMethod(promise, "then", resolve, reject); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case result := <-ch:
+		if !result.rejected {
+			t.Fatal(errors.New("promise resolved; want rejection"))
+		}
+		return result.value
+	case <-time.After(time.Second):
+		t.Fatal("promise did not settle")
+		return js.Undefined()
+	}
+}
+
+func awaitFulfilledPromise(t *testing.T, promise js.Value) js.Value {
+	t.Helper()
+	type result struct {
+		value    js.Value
+		rejected bool
+	}
+	ch := make(chan result, 1)
+	resolve := js.FuncOf(func(this js.Value, args []js.Value) any {
+		ch <- result{value: args[0]}
+		return nil
+	})
+	reject := js.FuncOf(func(this js.Value, args []js.Value) any {
+		ch <- result{value: args[0], rejected: true}
+		return nil
+	})
+	defer resolve.Release()
+	defer reject.Release()
+	if _, err := callJSMethod(promise, "then", resolve, reject); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case result := <-ch:
+		if result.rejected {
+			t.Fatalf("promise rejected with %v; want fulfillment", result.value)
+		}
+		return result.value
+	case <-time.After(time.Second):
+		t.Fatal("promise did not settle")
+		return js.Undefined()
+	}
+}

cmd/tsconnect/wasm/drive.go

@@ -30,6 +30,73 @@ type jsFileSystemForRemote struct {
 	fn js.Value
 }
 
+type driveResponse struct {
+	mu                   sync.Mutex
+	w                    http.ResponseWriter
+	doneCh               chan error
+	live                 bool
+	responseStarted      bool
+	testBeforeWriteCheck func()
+}
+
+func (r *driveResponse) isLive() bool {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	return r.live
+}
+
+func (r *driveResponse) writeHead(args []js.Value) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if !r.live || len(args) < 1 {
+		return
+	}
+	status := args[0].Int()
+	if len(args) > 1 && !args[1].IsUndefined() && !args[1].IsNull() {
+		for k, vs := range jsHeadersToGo(args[1]) {
+			for _, v := range vs {
+				r.w.Header().Add(k, v)
+			}
+		}
+	}
+	r.w.WriteHeader(status)
+	r.responseStarted = true
+}
+
+func (r *driveResponse) write(args []js.Value) {
+	if r.testBeforeWriteCheck != nil {
+		r.testBeforeWriteCheck()
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if !r.live || len(args) < 1 {
+		return
+	}
+	data := args[0]
+	buf := make([]byte, data.Get("length").Int())
+	js.CopyBytesToGo(buf, data)
+	r.responseStarted = true
+	if _, err := r.w.Write(buf); err != nil {
+		select {
+		case r.doneCh <- err:
+		default:
+		}
+		return
+	}
+	if f, ok := r.w.(http.Flusher); ok {
+		f.Flush()
+	}
+}
+
+func (r *driveResponse) finish(resultErr, contextErr error) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if resultErr != nil && contextErr == nil && !r.responseStarted {
+		http.Error(r.w, "drive handler failed", http.StatusInternalServerError)
+	}
+	r.live = false
+}
+
 func (fs *jsFileSystemForRemote) setHandler(fn js.Value) {
 	fs.mu.Lock()
 	fs.fn = fn
@@ -50,7 +117,8 @@ func (fs *jsFileSystemForRemote) Close() error { return nil }
 // response body back via write()/end() callbacks, so no full-body buffering
 // occurs regardless of file size.
 //
-// The call blocks until JS calls end() (or a write error occurs).
+// The call blocks until JS calls end(), a write or handler error occurs, or
+// the request context is cancelled.
 func (fs *jsFileSystemForRemote) ServeHTTPWithPerms(
 	perms drive.Permissions, w http.ResponseWriter, r *http.Request,
 ) {
@@ -58,14 +126,21 @@ func (fs *jsFileSystemForRemote) ServeHTTPWithPerms(
 	fn := fs.fn
 	fs.mu.RUnlock()
 
-	if fn.IsUndefined() || fn.IsNull() {
+	if fn.Type() != js.TypeFunction {
 		http.NotFound(w, r)
 		return
 	}
 
+	response := &driveResponse{w: w, doneCh: make(chan error, 1), live: true}
+
 	// readBodyChunk is exposed to JS as req.readBodyChunk().
 	// Each call returns a Promise<Uint8Array|null>: null signals EOF.
 	readBodyChunk := js.FuncOf(func(_ js.Value, _ []js.Value) any {
+		if !response.isLive() {
+			return makePromise(func() (any, error) {
+				return js.Null(), nil
+			})
+		}
 		return makePromise(func() (any, error) {
 			buf := make([]byte, 65536)
 			n, err := r.Body.Read(buf)
@@ -81,62 +156,54 @@ func (fs *jsFileSystemForRemote) ServeHTTPWithPerms(
 		})
 	})
 
-	// doneCh receives nil when JS calls end(), or a write error if Write fails.
-	doneCh := make(chan error, 1)
-
 	// writeHead sets response headers and status code. Must be called before write().
 	writeHead := js.FuncOf(func(_ js.Value, args []js.Value) any {
-		if len(args) < 1 {
+		response.writeHead(args)
-			return nil
-		}
-		status := args[0].Int()
-		if len(args) > 1 && !args[1].IsUndefined() && !args[1].IsNull() {
-			for k, vs := range jsHeadersToGo(args[1]) {
-				for _, v := range vs {
-					w.Header().Add(k, v)
-				}
-			}
-		}
-		w.WriteHeader(status)
 		return nil
 	})
 
 	// write streams a single response body chunk to the client.
 	write := js.FuncOf(func(_ js.Value, args []js.Value) any {
-		if len(args) < 1 {
+		response.write(args)
-			return nil
-		}
-		data := args[0]
-		buf := make([]byte, data.Get("length").Int())
-		js.CopyBytesToGo(buf, data)
-		if _, werr := w.Write(buf); werr != nil {
-			select {
-			case doneCh <- werr:
-			default:
-			}
-			return nil
-		}
-		if f, ok := w.(http.Flusher); ok {
-			f.Flush()
-		}
 		return nil
 	})
 
 	// end signals that the response is complete.
 	end := js.FuncOf(func(_ js.Value, _ []js.Value) any {
 		select {
-		case doneCh <- nil:
+		case response.doneCh <- nil:
 		default:
 		}
 		return nil
 	})
-
+	var fulfilled, rejected js.Func
-	defer func() {
+	var handlerSettled sync.Once
+	settleHandler := func(err error) {
+		handlerSettled.Do(func() {
+			select {
+			case response.doneCh <- err:
+			default:
+			}
 			readBodyChunk.Release()
 			writeHead.Release()
 			write.Release()
 			end.Release()
-	}()
+			fulfilled.Release()
+			rejected.Release()
+		})
+	}
+	fulfilled = js.FuncOf(func(_ js.Value, _ []js.Value) any {
+		settleHandler(nil)
+		return nil
+	})
+	rejected = js.FuncOf(func(_ js.Value, args []js.Value) any {
+		err := errors.New("JavaScript drive handler rejected")
+		if len(args) > 0 && args[0].Type() == js.TypeString {
+			err = fmt.Errorf("JavaScript drive handler rejected: %s", args[0].String())
+		}
+		settleHandler(err)
+		return nil
+	})
 
 	jsReq := map[string]any{
 		"method":        r.Method,
@@ -151,11 +218,28 @@ func (fs *jsFileSystemForRemote) ServeHTTPWithPerms(
 		"end":       end,
 	}
 
-	fn.Invoke(jsReq, jsRes, drivePermsToJS(perms))
+	result, handlerErr := callJSFunction(fn, jsReq, jsRes, drivePermsToJS(perms))
+	if handlerErr == nil {
+		if hasThen, err := hasJSFunctionProperty(result, "then"); err != nil {
+			handlerErr = err
+		} else if hasThen {
+			_, handlerErr = callJSMethod(result, "then", fulfilled, rejected)
+		} else {
+			settleHandler(nil)
+		}
+	}
+	if handlerErr != nil {
+		settleHandler(handlerErr)
+	}
 
-	// Block this goroutine until JS calls end() or a write error occurs.
+	var resultErr error
-	// The Go WASM scheduler yields back to JS while we wait.
+	select {
-	<-doneCh
+	case resultErr = <-response.doneCh:
+	case <-r.Context().Done():
+		resultErr = r.Context().Err()
+	}
+
+	response.finish(resultErr, r.Context().Err())
 }
 
 // drivePermsToJS converts drive.Permissions to a plain JS-friendly object.

cmd/tsconnect/wasm/taildrop.go

@@ -17,6 +17,7 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"sync"
 	"syscall/js"
 	"time"
 
@@ -109,7 +110,10 @@ func (i *jsIPN) sendFile(stableNodeID, filename string, stream js.Value, declare
 			return nil, fmt.Errorf("bogus peer URL: %w", err)
 		}
 
-		reader := stream.Call("getReader")
+		reader, err := callJSMethod(stream, "getReader")
+		if err != nil {
+			return nil, err
+		}
 		body := &jsStreamReader{reader: reader}
 
 		outgoing := ipn.OutgoingFile{
@@ -242,40 +246,100 @@ func wireTaildropFileOps(lb *ipnlocal.LocalBackend, jsObj js.Value) {
 // ReadableStreamDefaultReader. Each Read call awaits one reader.read() Promise,
 // using the channel+FuncOf pattern so Go blocks until JS delivers the chunk.
 type jsStreamReader struct {
+	mu                sync.Mutex
 	reader            js.Value
 	buf               []byte
 	done              bool
+	err               error
+	testCancelTimeout time.Duration
 }
 
+const jsStreamCancelTimeout = time.Second
+
 func (r *jsStreamReader) Read(p []byte) (int, error) {
+	r.mu.Lock()
+	if r.err != nil {
+		err := r.err
+		r.mu.Unlock()
+		return 0, err
+	}
 	if r.done {
+		r.mu.Unlock()
 		return 0, io.EOF
 	}
 	if len(r.buf) > 0 {
 		n := copy(p, r.buf)
 		r.buf = r.buf[n:]
+		r.mu.Unlock()
 		return n, nil
 	}
+	r.mu.Unlock()
 	type chunkResult struct {
 		data []byte
 		done bool
+		err  error
 	}
 	ch := make(chan chunkResult, 1)
+	settle := func(result chunkResult) {
+		select {
+		case ch <- result:
+		default:
+		}
+	}
 	thenFn := js.FuncOf(func(this js.Value, args []js.Value) any {
+		defer func() {
+			if recovered := recover(); recovered != nil {
+				settle(chunkResult{err: recoveredJSError(recovered)})
+			}
+		}()
+		if len(args) == 0 || args[0].Type() != js.TypeObject {
+			settle(chunkResult{err: errors.New("JavaScript stream read returned an invalid result")})
+			return nil
+		}
 		result := args[0]
-		if result.Get("done").Bool() {
+		done := result.Get("done")
-			ch <- chunkResult{done: true}
+		if done.Type() != js.TypeBoolean {
+			settle(chunkResult{err: errors.New("JavaScript stream read result has no boolean done property")})
+			return nil
+		}
+		if done.Bool() {
+			settle(chunkResult{done: true})
 		} else {
 			value := result.Get("value")
+			uint8Array := js.Global().Get("Uint8Array")
+			if value.Type() != js.TypeObject || uint8Array.Type() != js.TypeFunction || !value.InstanceOf(uint8Array) {
+				settle(chunkResult{err: errors.New("JavaScript stream read result is not a Uint8Array")})
+				return nil
+			}
 			b := make([]byte, value.Get("byteLength").Int())
 			js.CopyBytesToGo(b, value)
-			ch <- chunkResult{data: b}
+			settle(chunkResult{data: b})
 		}
 		return nil
 	})
+	rejectFn := js.FuncOf(func(this js.Value, args []js.Value) any {
+		err := errors.New("JavaScript stream read rejected")
+		if len(args) > 0 && args[0].Type() == js.TypeString {
+			err = fmt.Errorf("JavaScript stream read rejected: %s", args[0].String())
+		}
+		settle(chunkResult{err: err})
+		return nil
+	})
 	defer thenFn.Release()
-	r.reader.Call("read").Call("then", thenFn)
+	defer rejectFn.Release()
+	promise, err := callJSMethod(r.reader, "read")
+	if err != nil {
+		return 0, r.setError(err)
+	}
+	if _, err := callJSMethod(promise, "then", thenFn, rejectFn); err != nil {
+		return 0, r.setError(err)
+	}
 	result := <-ch
+	if result.err != nil {
+		return 0, r.setError(result.err)
+	}
+	r.mu.Lock()
+	defer r.mu.Unlock()
 	if result.done {
 		r.done = true
 		return 0, io.EOF
@@ -286,8 +350,72 @@ func (r *jsStreamReader) Read(p []byte) (int, error) {
 }
 
 func (r *jsStreamReader) Close() error {
-	r.reader.Call("cancel")
+	r.mu.Lock()
+	priorErr := r.err
+	r.mu.Unlock()
+	promise, err := callJSMethod(r.reader, "cancel")
+	if err != nil {
+		return r.setError(err)
+	}
+	if hasThen, err := hasJSFunctionProperty(promise, "then"); err != nil {
+		return r.setError(err)
+	} else if !hasThen {
+		r.mu.Lock()
+		r.done = true
+		r.mu.Unlock()
+		return priorErr
+	}
+	resultCh := make(chan error, 1)
+	settle := func(err error) {
+		select {
+		case resultCh <- err:
+		default:
+		}
+	}
+	resolveFn := js.FuncOf(func(this js.Value, args []js.Value) any {
+		settle(nil)
 		return nil
+	})
+	rejectFn := js.FuncOf(func(this js.Value, args []js.Value) any {
+		err := errors.New("JavaScript stream cancel rejected")
+		if len(args) > 0 && args[0].Type() == js.TypeString {
+			err = fmt.Errorf("JavaScript stream cancel rejected: %s", args[0].String())
+		}
+		settle(err)
+		return nil
+	})
+	defer resolveFn.Release()
+	defer rejectFn.Release()
+	if _, err := callJSMethod(promise, "then", resolveFn, rejectFn); err != nil {
+		return r.setError(err)
+	}
+	timeout := jsStreamCancelTimeout
+	if r.testCancelTimeout > 0 {
+		timeout = r.testCancelTimeout
+	}
+	timer := time.NewTimer(timeout)
+	defer timer.Stop()
+	select {
+	case err := <-resultCh:
+		if err != nil {
+			return r.setError(err)
+		}
+	case <-timer.C:
+		return r.setError(errors.New("JavaScript stream cancel timed out"))
+	}
+	r.mu.Lock()
+	r.done = true
+	r.mu.Unlock()
+	return priorErr
+}
+
+func (r *jsStreamReader) setError(err error) error {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if r.err == nil {
+		r.err = err
+	}
+	return r.err
 }
 
 // jsReadableStream wraps rc in a pull-based JS ReadableStream. Each pull call

cmd/tsconnect/wasm/wasm_js.go

@@ -432,6 +432,9 @@ type jsIPN struct {
 
 	terminate    func() // unblocks main() so the Go runtime can exit
 	shutdownOnce sync.Once
+	notifyMu     sync.Mutex
+	notifyCancel context.CancelFunc
+	notifyDone   <-chan struct{}
 }
 
 // funnelListenerEntry is the per-port state for routing Funnel connections to a listenTLS listener.
@@ -457,34 +460,62 @@ var jsMachineStatus = map[tailcfg.MachineStatus]string{
 	tailcfg.MachineInvalid:      "MachineInvalid",
 }
 
-func (i *jsIPN) run(jsCallbacks js.Value) {
+const jsIPNNotifyWatchMask = ipn.NotifyInitialState | ipn.NotifyInitialPrefs | ipn.NotifyPeerChanges | ipn.NotifyPeerPatches
-	notifyState := func(state ipn.State) {
-		jsCallbacks.Call("notifyState", jsIPNState[state])
-	}
-	notifyState(ipn.NoState)
 
-	i.lb.SetNotifyCallback(func(n ipn.Notify) {
+func notifyRefreshesNetMap(n ipn.Notify) bool {
-		// Panics in the notify callback are likely due to be due to bugs in
+	if n.SelfChange != nil || len(n.PeersChanged) > 0 || len(n.PeersRemoved) > 0 {
-		// this bridging module (as opposed to actual bugs in Tailscale) and
+		return true
-		// thus may be recoverable. Let the UI know, and allow the user to
-		// choose if they want to reload the page.
-		defer func() {
-			if r := recover(); r != nil {
-				fmt.Println("Panic recovered:", r)
-				jsCallbacks.Call("notifyPanicRecover", fmt.Sprint(r))
 	}
-		}()
+	for _, patch := range n.PeerChangedPatch {
-		log.Printf("NOTIFY: %+v", n)
+		if patch.Online != nil {
-		if n.State != nil {
+			return true
-			notifyState(*n.State)
 		}
-		if n.SelfChange != nil {
+	}
-			// Self changed: rebuild the JS-side NetMap snapshot. Peers
+	return false
-			// don't ride on the bus anymore, so fetch them on demand
+}
-			// from LocalBackend.
+
+func (i *jsIPN) watchNotifications(ctx context.Context, registered, done chan struct{}, notify func(ipn.Notify), refresh func()) {
+	defer close(done)
+	var registeredOnce sync.Once
+	for {
+		watchAdded := false
+		i.lb.WatchNotifications(ctx, jsIPNNotifyWatchMask, func() {
+			watchAdded = true
+			refresh()
+			registeredOnce.Do(func() { close(registered) })
+		}, func(n *ipn.Notify) bool {
+			if n.ErrMessage != nil {
+				log.Printf("IPN notification error: %s", *n.ErrMessage)
+			}
+			notify(*n)
+			return true
+		})
+		if ctx.Err() != nil {
+			return
+		}
+		if !watchAdded {
+			log.Printf("IPN notification watcher stopped before registration")
+			return
+		}
+		log.Printf("IPN notification watcher stopped; reconnecting")
+		timer := time.NewTimer(100 * time.Millisecond)
+		select {
+		case <-ctx.Done():
+			if !timer.Stop() {
+				<-timer.C
+			}
+			return
+		case <-timer.C:
+		}
+	}
+}
+
+func (i *jsIPN) refreshNetMap(jsCallbacks js.Value) {
 	nm := i.lb.NetMapWithPeers()
-			if nm != nil {
+	if nm == nil {
-				// Determine which address families we have, for peer peerAPI URL selection.
+		return
+	}
+
 	var selfHave4, selfHave6 bool
 	for _, a := range nm.GetAddresses().All() {
 		if !a.IsSingleIP() {
@@ -497,7 +528,6 @@ func (i *jsIPN) run(jsCallbacks js.Value) {
 		}
 	}
 
-				// Self peerAPI URL: own port as reported by LocalBackend.
 	selfPeerAPIURL := ""
 	for _, a := range nm.GetAddresses().All() {
 		if !a.IsSingleIP() {
@@ -524,24 +554,19 @@ func (i *jsIPN) run(jsCallbacks js.Value) {
 		Peers: mapSlice(nm.Peers, func(p tailcfg.NodeView) jsNetMapPeerNode {
 			name := p.Name()
 			if name == "" {
-							// In practice this should only happen for Hello.
 				name = p.Hostinfo().Hostname()
 			}
 			addrs := make([]string, p.Addresses().Len())
 			for idx, ap := range p.Addresses().All() {
 				addrs[idx] = ap.Addr().String()
 			}
-
-						// Peer peerAPI URL from the peer's advertised Services.
-						peerURL := buildPeerAPIURL(p, selfHave4, selfHave6)
-
 			return jsNetMapPeerNode{
 				jsNetMapNode: jsNetMapNode{
 					Name:       name,
 					Addresses:  addrs,
 					MachineKey: p.Machine().String(),
 					NodeKey:    p.Key().String(),
-								PeerAPIURL: peerURL,
+					PeerAPIURL: buildPeerAPIURL(p, selfHave4, selfHave6),
 					Services:   userServicesFromView(p.Hostinfo().Services()),
 				},
 				Online:              p.Online().Clone(),
@@ -557,7 +582,31 @@ func (i *jsIPN) run(jsCallbacks js.Value) {
 	} else {
 		log.Printf("Could not generate JSON netmap: %v", err)
 	}
+}
+
+func (i *jsIPN) run(jsCallbacks js.Value) {
+	notifyState := func(state ipn.State) {
+		jsCallbacks.Call("notifyState", jsIPNState[state])
 	}
+	notifyState(ipn.NoState)
+
+	notify := func(n ipn.Notify) {
+		// Panics in the notify callback are likely due to be due to bugs in
+		// this bridging module (as opposed to actual bugs in Tailscale) and
+		// thus may be recoverable. Let the UI know, and allow the user to
+		// choose if they want to reload the page.
+		defer func() {
+			if r := recover(); r != nil {
+				fmt.Println("Panic recovered:", r)
+				jsCallbacks.Call("notifyPanicRecover", fmt.Sprint(r))
+			}
+		}()
+		log.Printf("NOTIFY: %+v", n)
+		if n.State != nil {
+			notifyState(*n.State)
+		}
+		if notifyRefreshesNetMap(n) {
+			i.refreshNetMap(jsCallbacks)
 		}
 		if n.Prefs != nil && n.Prefs.Valid() {
 			jsCallbacks.Call("notifyExitNode", string(n.Prefs.ExitNodeID()))
@@ -605,9 +654,32 @@ func (i *jsIPN) run(jsCallbacks js.Value) {
 				log.Printf("could not marshal OutgoingFiles: %v", err)
 			}
 		}
-	})
+	}
+	refresh := func() {
+		defer func() {
+			if r := recover(); r != nil {
+				fmt.Println("Panic recovered:", r)
+				jsCallbacks.Call("notifyPanicRecover", fmt.Sprint(r))
+			}
+		}()
+		i.refreshNetMap(jsCallbacks)
+	}
+
+	registered := make(chan struct{})
+	notifyCtx, notifyCancel := context.WithCancel(context.Background())
+	notifyDone := make(chan struct{})
+	i.notifyMu.Lock()
+	i.notifyCancel = notifyCancel
+	i.notifyDone = notifyDone
+	i.notifyMu.Unlock()
+	go i.watchNotifications(notifyCtx, registered, notifyDone, notify, refresh)
 
 	go func() {
+		select {
+		case <-registered:
+		case <-notifyDone:
+			return
+		}
 		err := i.lb.Start(ipn.Options{
 			UpdatePrefs: &ipn.Prefs{
 				ControlURL:  i.controlURL,
@@ -646,6 +718,15 @@ func (i *jsIPN) logout() {
 func (i *jsIPN) shutdown() js.Value {
 	return makePromise(func() (any, error) {
 		i.shutdownOnce.Do(func() {
+			i.notifyMu.Lock()
+			notifyCancel, notifyDone := i.notifyCancel, i.notifyDone
+			i.notifyMu.Unlock()
+			if notifyCancel != nil {
+				notifyCancel()
+			}
+			if notifyDone != nil {
+				<-notifyDone
+			}
 			if i.lb != nil {
 				i.lb.Shutdown()
 			}
@@ -1591,19 +1672,83 @@ func makePromise(f func() (any, error)) js.Value {
 		resolve := args[0]
 		reject := args[1]
 		go func() {
+			defer func() {
+				if recovered := recover(); recovered != nil {
+					rejectJSError(reject, recoveredJSError(recovered))
+				}
+			}()
 			if res, err := f(); err == nil {
 				resolve.Invoke(res)
 			} else {
-				reject.Invoke(err.Error())
+				rejectJSError(reject, err)
 			}
 		}()
 		return nil
 	})
+	defer handler.Release()
 
 	promiseConstructor := js.Global().Get("Promise")
 	return promiseConstructor.New(handler)
 }
 
+func callJSFunction(fn js.Value, args ...any) (ret js.Value, err error) {
+	if fn.Type() != js.TypeFunction {
+		return js.Undefined(), fmt.Errorf("expected JavaScript function, got %s", fn.Type())
+	}
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = recoveredJSError(recovered)
+		}
+	}()
+	return fn.Invoke(args...), nil
+}
+
+func callJSMethod(receiver js.Value, method string, args ...any) (ret js.Value, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = recoveredJSError(recovered)
+		}
+	}()
+	if receiver.Type() != js.TypeObject && receiver.Type() != js.TypeFunction {
+		return js.Undefined(), fmt.Errorf("cannot call JavaScript method %q on %s", method, receiver.Type())
+	}
+	if receiver.Get(method).Type() != js.TypeFunction {
+		return js.Undefined(), fmt.Errorf("JavaScript property %q is not a function", method)
+	}
+	return receiver.Call(method, args...), nil
+}
+
+func hasJSFunctionProperty(receiver js.Value, property string) (ok bool, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			err = recoveredJSError(recovered)
+		}
+	}()
+	if receiver.Type() != js.TypeObject && receiver.Type() != js.TypeFunction {
+		return false, nil
+	}
+	return receiver.Get(property).Type() == js.TypeFunction, nil
+}
+
+func recoveredJSError(recovered any) error {
+	switch recovered := recovered.(type) {
+	case js.Error:
+		return recovered
+	case *js.ValueError:
+		return recovered
+	default:
+		panic(recovered)
+	}
+}
+
+func rejectJSError(reject js.Value, err error) {
+	if jsErr, ok := err.(js.Error); ok {
+		reject.Invoke(jsErr.Value)
+		return
+	}
+	reject.Invoke(err.Error())
+}
+
 const logPolicyStateKey = "log-policy"
 
 func getOrCreateLogPolicyConfig(state ipn.StateStore) *logpolicy.Config {

cmd/tsconnect/wasmbuild/printtags/main.go

@@ -0,0 +1,14 @@
+// Copyright (c) Tailscale Inc & contributors
+// SPDX-License-Identifier: BSD-3-Clause
+
+package main
+
+import (
+	"fmt"
+
+	"tailscale.com/cmd/tsconnect/wasmbuild"
+)
+
+func main() {
+	fmt.Print(wasmbuild.Tags())
+}