tsconnect: harden JS bridge callbacks

Contain synchronous and asynchronous JS failures, make Drive cancellation fence response callbacks, and keep netmap snapshots current across peer changes and watcher restarts.

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

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,97 @@ 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
 }
 
 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() {
-			ch <- chunkResult{done: true}
+		done := result.Get("done")
+		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 +347,61 @@ func (r *jsStreamReader) Read(p []byte) (int, error) {
 }
 
 func (r *jsStreamReader) Close() error {
-	r.reader.Call("cancel")
-	return nil
+	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)
+	}
+	if err := <-resultCh; err != nil {
+		return r.setError(err)
+	}
+	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